test/chore: Bandit findings on PR #3350 marked with reasons (ftplib in tests, migration SQL built from module constants), a misplaced nosec moved to its SQL line; layer timelapse tests use tmp_path instead of a shared /tmp/test_bambuddy

backend/app/core/database.py

@@ -448,9 +448,9 @@ async def _migrate_sqlite_autoincrement(conn) -> None:
             await conn.execute(text(f"DROP TABLE IF EXISTS {copy}"))
             await conn.execute(text(new_sql))
             names = ", ".join(f'"{r[1]}"' for r in await conn.execute(text(f"PRAGMA table_info({table})")))
-            await conn.execute(text(f"INSERT INTO {copy} ({names}) SELECT {names} FROM {table}"))
-            original = (await conn.execute(text(f"SELECT count(*) FROM {table}"))).scalar_one()
-            copied = (await conn.execute(text(f"SELECT count(*) FROM {copy}"))).scalar_one()
+            await conn.execute(text(f"INSERT INTO {copy} ({names}) SELECT {names} FROM {table}"))  # nosec B608 -- table from AUTOINCREMENT_TABLES, columns from its own PRAGMA
+            original = (await conn.execute(text(f"SELECT count(*) FROM {table}"))).scalar_one()  # nosec B608 -- table from AUTOINCREMENT_TABLES, columns from its own PRAGMA
+            copied = (await conn.execute(text(f"SELECT count(*) FROM {copy}"))).scalar_one()  # nosec B608 -- table from AUTOINCREMENT_TABLES, columns from its own PRAGMA
             if copied != original:
                 raise RuntimeError(f"{table} AUTOINCREMENT rebuild: {original} rows, {copied} copied")
             await conn.execute(text(f"DROP TABLE {table}"))
@@ -5985,7 +5985,7 @@ async def _migrate_pg_columns_to_model(conn) -> None:
         extra = ""
         if (table, column) == ("oidc_providers", "email_claim"):
             extra = ", auto_link_existing_accounts = false"
-        filled = await conn.execute(text(f"UPDATE {table} SET {column} = {fill}{extra} WHERE {column} IS NULL"))
+        filled = await conn.execute(text(f"UPDATE {table} SET {column} = {fill}{extra} WHERE {column} IS NULL"))  # nosec B608 -- table/column from _PG_COLUMNS_SET_FROM_MODEL, fill from the model default
         if filled.rowcount:
             logger.info("Filled %d NULL %s.%s value(s) with %s", filled.rowcount, table, column, fill)
         await _safe_execute(conn, f"ALTER TABLE {table} ALTER COLUMN {column} SET NOT NULL")
@@ -6052,13 +6052,13 @@ async def _migrate_pg_column_types(conn) -> None:
         if current != old_type:
             continue
         if new_type == "JSON":
-            rows = await conn.execute(text(f"SELECT id, {column} FROM {table} WHERE {column} IS NOT NULL"))
+            rows = await conn.execute(text(f"SELECT id, {column} FROM {table} WHERE {column} IS NOT NULL"))  # nosec B608 -- table/column from _PG_COLUMN_TYPE_FIXES
             for row_id, value in rows.all():
                 try:
                     json.loads(value, parse_constant=_reject_non_json_constant)
                 except ValueError:
                     logger.warning("Clearing %s.%s of row %s: not valid JSON", table, column, row_id)
-                    await conn.execute(text(f"UPDATE {table} SET {column} = NULL WHERE id = :id"), {"id": row_id})
+                    await conn.execute(text(f"UPDATE {table} SET {column} = NULL WHERE id = :id"), {"id": row_id})  # nosec B608 -- table/column from _PG_COLUMN_TYPE_FIXES; id bound
         # Float to text gives the shortest exact decimal only with
         # extra_float_digits >= 1, PostgreSQL's default since 12, which a server,
         # database or role setting can lower: at 0 text keeps 6 digits.
@@ -6315,7 +6315,7 @@ async def _migrate_add_missing_set_null_fks(conn) -> None:
             continue
         cleared = await conn.execute(
             text(
-                f"UPDATE {table} SET {column} = NULL WHERE {column} IS NOT NULL "
+                f"UPDATE {table} SET {column} = NULL WHERE {column} IS NOT NULL "  # nosec B608 -- table/column/parent from _MISSING_SET_NULL_FKS
                 f"AND NOT EXISTS (SELECT 1 FROM {parent} WHERE {parent}.id = {table}.{column})"
             )
         )
@@ -6975,8 +6975,8 @@ async def _migrate_normalize_printer_locations(conn) -> None:
             )
             await conn.execute(
                 text(
-                    f"UPDATE {table} SET {column} = TRIM({column}) WHERE {column} IS NOT NULL AND {column} <> TRIM({column})"
-                )  # noqa: S608  # nosec B608 — fixed identifiers
+                    f"UPDATE {table} SET {column} = TRIM({column}) WHERE {column} IS NOT NULL AND {column} <> TRIM({column})"  # noqa: S608  # nosec B608 — fixed identifiers
+                )
             )
 
 

backend/tests/integration/test_pipeline_runs_api.py

@@ -1420,7 +1420,7 @@ class TestJobKeepsItsFinalStatus:
         )
         connection = await db_session.connection()
         # Ended before jobs stored it: written past the session hooks.
-        await connection.exec_driver_sql(f"UPDATE print_queue SET status = 'aborted' WHERE id = {int(item_id)}")
+        await connection.exec_driver_sql(f"UPDATE print_queue SET status = 'aborted' WHERE id = {int(item_id)}")  # nosec B608 -- int() id
         await db_session.commit()
 
         await _backfill_pipeline_job_final_status(await db_session.connection())

backend/tests/integration/test_print_queue_api.py

@@ -62,7 +62,7 @@ class TestPrintQueueAPI:
             defaults = {
                 "filename": f"test_print_{counter}.3mf",
                 "print_name": f"Test Print {counter}",
-                "file_path": f"/tmp/test_print_{counter}.3mf",
+                "file_path": f"/tmp/test_print_{counter}.3mf",  # nosec B108 -- a stored path, never opened
                 "file_size": 1024,
                 "content_hash": f"testhash{counter:08d}",
                 "status": "completed",
@@ -1033,7 +1033,7 @@ class TestQueueStartEndpoint:
             defaults = {
                 "filename": f"test_print_{counter}.3mf",
                 "print_name": f"Test Print {counter}",
-                "file_path": f"/tmp/test_print_{counter}.3mf",
+                "file_path": f"/tmp/test_print_{counter}.3mf",  # nosec B108 -- a stored path, never opened
                 "file_size": 1024,
                 "content_hash": f"testhash{counter:08d}",
                 "status": "completed",
@@ -1381,7 +1381,7 @@ class TestQueueCancelEndpoint:
             defaults = {
                 "filename": "cancel_test.3mf",
                 "print_name": "Cancel Test Print",
-                "file_path": "/tmp/cancel_test.3mf",
+                "file_path": "/tmp/cancel_test.3mf",  # nosec B108 -- a stored path, never opened
                 "file_size": 1024,
                 "content_hash": "cancelhash001",
                 "status": "completed",
@@ -1700,7 +1700,7 @@ class TestBulkUpdateEndpoint:
             defaults = {
                 "filename": f"bulk_test_{counter}.3mf",
                 "print_name": f"Bulk Test Print {counter}",
-                "file_path": f"/tmp/bulk_test_{counter}.3mf",
+                "file_path": f"/tmp/bulk_test_{counter}.3mf",  # nosec B108 -- a stored path, never opened
                 "file_size": 1024,
                 "content_hash": f"bulkhash{counter:04d}",
                 "status": "completed",
@@ -1981,7 +1981,7 @@ class TestTargetLocationFeature:
             defaults = {
                 "filename": f"location_test_{counter}.3mf",
                 "print_name": f"Location Test Print {counter}",
-                "file_path": f"/tmp/location_test_{counter}.3mf",
+                "file_path": f"/tmp/location_test_{counter}.3mf",  # nosec B108 -- a stored path, never opened
                 "file_size": 1024,
                 "content_hash": f"lochash{counter:08d}",
                 "status": "completed",
@@ -2310,7 +2310,7 @@ class TestAbortedStatusNormalisation:
             defaults = {
                 "filename": f"abort_test_{counter}.3mf",
                 "print_name": f"Abort Test Print {counter}",
-                "file_path": f"/tmp/abort_test_{counter}.3mf",
+                "file_path": f"/tmp/abort_test_{counter}.3mf",  # nosec B108 -- a stored path, never opened
                 "file_size": 1024,
                 "content_hash": f"aborthash{counter:06d}",
                 "status": "completed",

backend/tests/integration/test_virtual_printer_api.py

@@ -136,7 +136,7 @@ class TestPendingUploadsAPI:
 
             upload = PendingUpload(
                 filename=filename,
-                file_path=f"/tmp/{filename}",
+                file_path=f"/tmp/{filename}",  # nosec B108 -- a stored path, never opened
                 file_size=1024,
                 source_ip="192.168.1.100",
                 status="pending",

backend/tests/unit/services/test_ftp_download_deadlines.py

@@ -136,7 +136,7 @@ def _client_with(error: Exception) -> bambu_ftp.BambuFTPClient:
 
 
 def test_a_missing_path_is_told_apart_when_asked():
-    import ftplib
+    import ftplib  # nosec B402 -- tests construct real ftplib error types
 
     missing = ftplib.error_perm("550 No such file")
     with pytest.raises(bambu_ftp.FileNotOnPrinterError):
@@ -148,7 +148,7 @@ def test_a_missing_path_is_told_apart_when_asked():
 
 
 def test_other_ftp_errors_are_failures_not_missing():
-    import ftplib
+    import ftplib  # nosec B402 -- tests construct real ftplib error types
 
     assert _client_with(ftplib.error_perm("530 Not logged in")).download_file("/x", report_missing=True) is None
     assert _client_with(OSError("reset")).download_file("/x", report_missing=True) is None
@@ -219,7 +219,7 @@ async def test_a_printer_file_past_its_deadline_is_a_timeout(monkeypatch, tmp_pa
 def test_a_size_refusal_is_left_to_retr():
     """A 550 to SIZE (no such command, a file still being written) does not
     make a file missing that RETR then serves."""
-    import ftplib
+    import ftplib  # nosec B402 -- tests construct real ftplib error types
     from unittest.mock import MagicMock
 
     client = bambu_ftp.BambuFTPClient("10.0.0.5", "code")

backend/tests/unit/services/test_layer_timelapse.py

@@ -15,7 +15,7 @@ import pytest
 class TestTimelapseSessionManagement:
     """Tests for timelapse session lifecycle."""
 
-    def test_start_session_creates_new_session(self):
+    def test_start_session_creates_new_session(self, tmp_path):
         """Verify start_session creates and registers a new session."""
         from backend.app.services.layer_timelapse import (
             _active_sessions,
@@ -28,7 +28,7 @@ class TestTimelapseSessionManagement:
         _active_sessions.clear()
 
         with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
-            mock_settings.base_dir = Path("/tmp/test_bambuddy")
+            mock_settings.base_dir = tmp_path
 
             session = start_session(
                 printer_id=1,
@@ -52,7 +52,7 @@ class TestTimelapseSessionManagement:
             # Cleanup
             cancel_session(1)
 
-    def test_start_session_cancels_existing(self):
+    def test_start_session_cancels_existing(self, tmp_path):
         """Verify starting a new session cancels any existing session."""
         from backend.app.services.layer_timelapse import (
             _active_sessions,
@@ -64,7 +64,7 @@ class TestTimelapseSessionManagement:
         _active_sessions.clear()
 
         with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
-            mock_settings.base_dir = Path("/tmp/test_bambuddy")
+            mock_settings.base_dir = tmp_path
 
             # Start first session
             session1 = start_session(1, 100, "http://cam1/", "mjpeg")
@@ -96,7 +96,7 @@ class TestTimelapseSessionManagement:
         result = get_session(999)
         assert result is None
 
-    def test_cancel_session_removes_and_cleans_up(self):
+    def test_cancel_session_removes_and_cleans_up(self, tmp_path):
         """Verify cancel_session removes session and cleans up."""
         from backend.app.services.layer_timelapse import (
             _active_sessions,
@@ -108,7 +108,7 @@ class TestTimelapseSessionManagement:
         _active_sessions.clear()
 
         with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
-            mock_settings.base_dir = Path("/tmp/test_bambuddy")
+            mock_settings.base_dir = tmp_path
 
             session = start_session(1, 100, "http://cam/", "mjpeg")
 
@@ -135,14 +135,14 @@ class TestTimelapseSessionManagement:
 class TestTimelapseSession:
     """Tests for TimelapseSession class."""
 
-    def test_session_id_format(self):
+    def test_session_id_format(self, tmp_path):
         """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")
+            mock_settings.base_dir = tmp_path
 
             session = TimelapseSession(
                 printer_id=1,
@@ -184,12 +184,12 @@ class TestLayerChangeLogic:
     """Tests for layer change capture logic."""
 
     @pytest.mark.asyncio
-    async def test_capture_layer_only_on_increase(self):
+    async def test_capture_layer_only_on_increase(self, tmp_path):
         """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")
+            mock_settings.base_dir = tmp_path
 
             with patch.object(Path, "mkdir"):
                 session = TimelapseSession(1, 100, "http://test/", "mjpeg")
@@ -224,12 +224,12 @@ class TestLayerChangeLogic:
                         assert session.frame_count == 2
 
     @pytest.mark.asyncio
-    async def test_capture_layer_handles_failed_capture(self):
+    async def test_capture_layer_handles_failed_capture(self, tmp_path):
         """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")
+            mock_settings.base_dir = tmp_path
 
             with patch.object(Path, "mkdir"):
                 session = TimelapseSession(1, 100, "http://test/", "mjpeg")
@@ -344,7 +344,7 @@ class TestOnLayerChange:
     """Tests for the on_layer_change callback."""
 
     @pytest.mark.asyncio
-    async def test_on_layer_change_captures_when_session_exists(self):
+    async def test_on_layer_change_captures_when_session_exists(self, tmp_path):
         """Verify on_layer_change triggers capture when session exists."""
         from backend.app.services.layer_timelapse import (
             _active_sessions,
@@ -356,7 +356,7 @@ class TestOnLayerChange:
         _active_sessions.clear()
 
         with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
-            mock_settings.base_dir = Path("/tmp/test")
+            mock_settings.base_dir = tmp_path
 
             with patch.object(Path, "mkdir"):
                 session = start_session(1, 100, "http://test/", "mjpeg")