added tests for finance/cost center functionality

backend/tests/integration/test_background_dispatch_api.py

@@ -3,9 +3,23 @@
 from unittest.mock import AsyncMock, patch
 
 import pytest
+from fastapi import HTTPException
 from httpx import AsyncClient
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
-from backend.app.services.background_dispatch import DispatchEnqueueRejected
+from backend.app.models.finance import BudgetReservation, CostCenter
+from backend.app.models.settings import Settings
+from backend.app.services.background_dispatch import BackgroundDispatchService, DispatchEnqueueRejected
+
+
+async def enable_billing(db_session):
+    setting = await db_session.scalar(select(Settings).where(Settings.key == "billing_enabled"))
+    if setting is None:
+        db_session.add(Settings(key="billing_enabled", value="true"))
+    else:
+        setting.value = "true"
+    await db_session.commit()
 
 
 class TestBackgroundDispatchArchivesAPI:
@@ -53,6 +67,80 @@ class TestBackgroundDispatchArchivesAPI:
         assert kwargs["archive_name"].endswith("• Plate 2")
         assert kwargs["options"]["plate_id"] == 2
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_reprint_forwards_cost_center_id(
+        self, async_client: AsyncClient, archive_factory, printer_factory, db_session, tmp_path
+    ):
+        """Reprint endpoint forwards cost_center_id into the dispatch options."""
+        await enable_billing(db_session)
+        printer = await printer_factory()
+        archive = await archive_factory(
+            printer.id,
+            filename="widget-cost.gcode.3mf",
+            file_path="archives/test/widget-cost.gcode.3mf",
+        )
+
+        archive_file = tmp_path / archive.file_path
+        archive_file.parent.mkdir(parents=True, exist_ok=True)
+        archive_file.write_bytes(b"3mf-data")
+        cost_center = CostCenter(name="Reprint Cost Center", is_active=True, is_private=False, monthly_budget=10.0)
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+
+        with (
+            patch("backend.app.api.routes.archives.settings.base_dir", tmp_path),
+            patch("backend.app.services.printer_manager.printer_manager.is_connected", return_value=True),
+            patch(
+                "backend.app.services.background_dispatch.background_dispatch.dispatch_reprint_archive",
+                new=AsyncMock(return_value={"dispatch_job_id": 16, "dispatch_position": 1}),
+            ) as mock_dispatch,
+        ):
+            response = await async_client.post(
+                f"/api/v1/archives/{archive.id}/reprint?printer_id={printer.id}",
+                json={"plate_id": 2, "cost_center_id": cost_center.id, "estimated_cost": 1.25},
+            )
+
+        assert response.status_code == 200
+        mock_dispatch.assert_awaited_once()
+        assert mock_dispatch.await_args.kwargs["options"]["cost_center_id"] == cost_center.id
+        assert mock_dispatch.await_args.kwargs["options"]["estimated_cost"] == 1.25
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_reprint_requires_cost_center_when_billing_enabled(
+        self, async_client: AsyncClient, archive_factory, printer_factory, db_session, tmp_path
+    ):
+        await enable_billing(db_session)
+        printer = await printer_factory()
+        archive = await archive_factory(
+            printer.id,
+            filename="widget-missing-cost-center.gcode.3mf",
+            file_path="archives/test/widget-missing-cost-center.gcode.3mf",
+        )
+
+        archive_file = tmp_path / archive.file_path
+        archive_file.parent.mkdir(parents=True, exist_ok=True)
+        archive_file.write_bytes(b"3mf-data")
+
+        with (
+            patch("backend.app.api.routes.archives.settings.base_dir", tmp_path),
+            patch("backend.app.services.printer_manager.printer_manager.is_connected", return_value=True),
+            patch(
+                "backend.app.services.background_dispatch.background_dispatch.dispatch_reprint_archive",
+                new=AsyncMock(return_value={"dispatch_job_id": 17, "dispatch_position": 1}),
+            ) as mock_dispatch,
+        ):
+            response = await async_client.post(
+                f"/api/v1/archives/{archive.id}/reprint?printer_id={printer.id}",
+                json={"plate_id": 2, "estimated_cost": 1.25},
+            )
+
+        assert response.status_code == 400
+        assert "Cost center is required" in response.json()["detail"]
+        mock_dispatch.assert_not_awaited()
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_reprint_returns_409_when_enqueue_rejected(
@@ -150,6 +238,42 @@ class TestBackgroundDispatchLibraryAPI:
         assert kwargs["filename"].endswith("• Plate 4")
         assert kwargs["options"]["plate_id"] == 4
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_library_print_forwards_cost_center_id(
+        self, async_client: AsyncClient, library_file_factory, printer_factory, db_session, tmp_path
+    ):
+        """Library print endpoint forwards cost_center_id into the dispatch options."""
+        await enable_billing(db_session)
+        printer = await printer_factory()
+        lib_file = await library_file_factory(filename="library-cost.gcode.3mf")
+
+        disk_path = tmp_path / lib_file.file_path
+        disk_path.parent.mkdir(parents=True, exist_ok=True)
+        disk_path.write_bytes(b"library data")
+        cost_center = CostCenter(name="Library Cost Center", is_active=True, is_private=False, monthly_budget=10.0)
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+
+        with (
+            patch("backend.app.api.routes.library.app_settings.base_dir", tmp_path),
+            patch("backend.app.services.printer_manager.printer_manager.is_connected", return_value=True),
+            patch(
+                "backend.app.services.background_dispatch.background_dispatch.dispatch_print_library_file",
+                new=AsyncMock(return_value={"dispatch_job_id": 22, "dispatch_position": 2}),
+            ) as mock_dispatch,
+        ):
+            response = await async_client.post(
+                f"/api/v1/library/files/{lib_file.id}/print?printer_id={printer.id}",
+                json={"plate_id": 4, "cost_center_id": cost_center.id, "estimated_cost": 1.5},
+            )
+
+        assert response.status_code == 200
+        mock_dispatch.assert_awaited_once()
+        assert mock_dispatch.await_args.kwargs["options"]["cost_center_id"] == cost_center.id
+        assert mock_dispatch.await_args.kwargs["options"]["estimated_cost"] == 1.5
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_library_print_returns_409_when_enqueue_rejected(
@@ -304,3 +428,62 @@ class TestBackgroundDispatchCancelAPI:
 
         assert response.status_code == 404
         assert response.json()["detail"] == "Dispatch job not found"
+
+
+class TestBackgroundDispatchBudgetReservations:
+    """Tests for persisted budget reservations on background dispatch enqueue."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_background_dispatch_persists_budget_reservation_and_blocks_oversubscribe(
+        self, printer_factory, db_session, test_engine
+    ):
+        await enable_billing(db_session)
+        printer_one = await printer_factory()
+        printer_two = await printer_factory()
+        cost_center = CostCenter(name="Dispatch Reservation CC", is_active=True, is_private=False, monthly_budget=2.0)
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+
+        service = BackgroundDispatchService()
+        test_async_session = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+        with (
+            patch("backend.app.services.background_dispatch.async_session", test_async_session),
+            patch("backend.app.services.printer_manager.printer_manager.get_status", return_value=None),
+        ):
+            first = await service.dispatch_reprint_archive(
+                archive_id=100,
+                archive_name="first.gcode.3mf",
+                printer_id=printer_one.id,
+                printer_name=printer_one.name,
+                options={"cost_center_id": cost_center.id, "estimated_cost": 1.5},
+                requested_by_user_id=None,
+                requested_by_username=None,
+            )
+
+            with pytest.raises(HTTPException) as exc:
+                await service.dispatch_reprint_archive(
+                    archive_id=101,
+                    archive_name="second.gcode.3mf",
+                    printer_id=printer_two.id,
+                    printer_name=printer_two.name,
+                    options={"cost_center_id": cost_center.id, "estimated_cost": 1.0},
+                    requested_by_user_id=None,
+                    requested_by_username=None,
+                )
+
+        assert exc.value.status_code == 400
+        assert "exceeds available cost center budget" in exc.value.detail
+
+        reservation = await db_session.scalar(
+            select(BudgetReservation).where(
+                BudgetReservation.source_type == "background_dispatch",
+                BudgetReservation.source_id == first["dispatch_job_id"],
+            )
+        )
+        assert reservation is not None
+        assert reservation.status == "active"
+        assert reservation.cost_center_id == cost_center.id
+        assert reservation.amount == 1.5
+        assert reservation.print_archive_id == 100

backend/tests/integration/test_finance_api.py

@@ -0,0 +1,799 @@
+"""Integration tests for the finance/billing API."""
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from backend.app.core.auth import get_password_hash
+from backend.app.models.archive import PrintArchive
+from backend.app.models.finance import CostCenter, UserWallet, WalletTransaction
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+
+
+class TestFinanceAPI:
+    @pytest.fixture
+    async def admin_user(self, db_session):
+        user = User(
+            username="finance-admin",
+            email="finance-admin@example.com",
+            password_hash=get_password_hash("AdminPass1!"),
+            role="admin",
+            is_active=True,
+        )
+        db_session.add(user)
+        await db_session.commit()
+        await db_session.refresh(user)
+        return user
+
+    @pytest.fixture
+    async def auth_headers(self, async_client: AsyncClient, db_session, admin_user):
+        db_session.add(Settings(key="auth_enabled", value="true"))
+        db_session.add(Settings(key="advanced_auth_enabled", value="false"))
+        # Ensure billing is enabled for finance integration tests
+        existing = await db_session.scalar(select(Settings).where(Settings.key == "billing_enabled"))
+        if existing is None:
+            db_session.add(Settings(key="billing_enabled", value="true"))
+        else:
+            existing.value = "true"
+        await db_session.commit()
+
+        response = await async_client.post(
+            "/api/v1/auth/login",
+            json={"username": admin_user.username, "password": "AdminPass1!"},
+        )
+        assert response.status_code == 200
+        return {"Authorization": f"Bearer {response.json()['access_token']}"}
+
+    async def _enable_basic_user_creation(self, db_session):
+        return None
+
+    async def _create_user_via_api(self, async_client: AsyncClient, auth_headers: dict[str, str], username: str):
+        response = await async_client.post(
+            "/api/v1/users",
+            json={
+                "username": username,
+                "password": "Regularpass1!",
+                "email": f"{username}@example.com",
+                "role": "user",
+            },
+            headers=auth_headers,
+        )
+        assert response.status_code == 201
+        return response.json()
+
+    async def _login_user(self, async_client: AsyncClient, username: str) -> dict[str, str]:
+        response = await async_client.post(
+            "/api/v1/auth/login",
+            json={"username": username, "password": "Regularpass1!"},
+        )
+        assert response.status_code == 200
+        return {"Authorization": f"Bearer {response.json()['access_token']}"}
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_create_cost_center_assign_member_and_list_mine(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        db_session,
+    ):
+        await self._enable_basic_user_creation(db_session)
+        created_user = await self._create_user_via_api(async_client, auth_headers, "carol")
+        user_headers = await self._login_user(async_client, "carol")
+
+        create_response = await async_client.post(
+            "/api/v1/finance/cost-centers",
+            json={
+                "name": "Shared Lab",
+                "monthly_budget": 120.0,
+                "total_budget": 999.0,
+                "is_active": True,
+            },
+            headers=auth_headers,
+        )
+
+        assert create_response.status_code == 200
+        shared_center = create_response.json()
+        assert shared_center["name"] == "Shared Lab"
+        assert shared_center["monthly_budget"] == 120.0
+        assert shared_center["total_budget"] is None
+        assert shared_center["budget_mode"] == "monthly"
+
+        member_response = await async_client.post(
+            f"/api/v1/finance/cost-centers/{shared_center['id']}/members",
+            json={"user_id": created_user["id"], "can_print": False},
+            headers=auth_headers,
+        )
+
+        assert member_response.status_code == 200
+        assert member_response.json()["user_id"] == created_user["id"]
+        assert member_response.json()["can_print"] is False
+
+        mine_response = await async_client.get("/api/v1/finance/cost-centers/mine", headers=user_headers)
+        assert mine_response.status_code == 200
+        mine_names = {center["name"] for center in mine_response.json()}
+        assert "carol" in mine_names
+        assert "Shared Lab" in mine_names
+
+        detail_response = await async_client.get(
+            f"/api/v1/finance/cost-centers/{shared_center['id']}", headers=auth_headers
+        )
+        assert detail_response.status_code == 200
+        detail = detail_response.json()
+        assert len(detail["members"]) == 1
+        assert detail["members"][0]["user_id"] == created_user["id"]
+
+        remove_response = await async_client.delete(
+            f"/api/v1/finance/cost-centers/{shared_center['id']}/members/{created_user['id']}",
+            headers=auth_headers,
+        )
+        assert remove_response.status_code == 200
+
+        mine_after_remove = await async_client.get("/api/v1/finance/cost-centers/mine", headers=user_headers)
+        assert mine_after_remove.status_code == 200
+        assert {center["name"] for center in mine_after_remove.json()} == {"carol"}
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_wallet_adjustments_and_transaction_ledger_rebuild(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        admin_user,
+        db_session,
+    ):
+        """Test cost-center and personal transaction handling.
+
+        Cost-center transactions affect cost-center balance only.
+        Personal transactions (no cost_center_id) affect user wallet.
+        """
+        await self._enable_basic_user_creation(db_session)
+        created_user = await self._create_user_via_api(async_client, auth_headers, "dave")
+
+        shared_center = await db_session.scalar(
+            select(CostCenter).where(CostCenter.owner_user_id == created_user["id"], CostCenter.is_private.is_(True))
+        )
+        assert shared_center is not None
+
+        # Deposit to cost center: affects cost-center balance, not user wallet
+        deposit = await async_client.post(
+            f"/api/v1/finance/users/{created_user['id']}/deposit",
+            json={"amount": 25.0, "description": "Initial CC top-up", "cost_center_id": shared_center.id},
+            headers=auth_headers,
+        )
+        assert deposit.status_code == 200
+        assert deposit.json()["transaction"]["cost_center_id"] == shared_center.id
+        assert deposit.json()["transaction"]["balance_after"] == 25.0  # CC balance
+        assert deposit.json()["balance"]["balance"] == 25.0  # Response shows CC balance
+
+        # Withdraw from cost center: affects cost-center balance only
+        withdraw = await async_client.post(
+            f"/api/v1/finance/users/{created_user['id']}/withdraw",
+            json={"amount": 5.0, "description": "CC Usage", "cost_center_id": shared_center.id},
+            headers=auth_headers,
+        )
+        assert withdraw.status_code == 200
+        assert withdraw.json()["transaction"]["amount"] == -5.0
+        assert withdraw.json()["transaction"]["balance_after"] == 20.0  # CC balance after withdraw
+        assert withdraw.json()["balance"]["balance"] == 20.0  # Response shows CC balance
+
+        # Personal deposit: affects user wallet
+        personal_deposit = await async_client.post(
+            f"/api/v1/finance/users/{created_user['id']}/deposit",
+            json={"amount": 30.0, "description": "Personal top-up", "cost_center_id": None},
+            headers=auth_headers,
+        )
+        assert personal_deposit.status_code == 200
+        assert personal_deposit.json()["transaction"]["cost_center_id"] is None
+        assert personal_deposit.json()["transaction"]["balance_after"] == 30.0  # Personal balance
+        assert personal_deposit.json()["balance"]["balance"] == 30.0  # User wallet updated
+
+        transactions_response = await async_client.get(
+            f"/api/v1/finance/users/{created_user['id']}/transactions", headers=auth_headers
+        )
+        assert transactions_response.status_code == 200
+        transactions = transactions_response.json()
+        assert len(transactions) == 3
+        cc_txs = [tx for tx in transactions if tx["cost_center_id"] == shared_center.id]
+        personal_txs = [tx for tx in transactions if tx["cost_center_id"] is None]
+        assert len(cc_txs) == 2
+        assert len(personal_txs) == 1
+
+        balance_response = await async_client.get(
+            f"/api/v1/finance/users/{created_user['id']}/balance", headers=auth_headers
+        )
+        assert balance_response.status_code == 200
+        assert balance_response.json()["balance"] == 30.0  # Only personal balance
+
+        # Delete personal transaction, user wallet should decrease
+        personal_tx = next(tx for tx in transactions if tx["cost_center_id"] is None)
+        delete_response = await async_client.delete(
+            f"/api/v1/finance/transactions/{personal_tx['id']}", headers=auth_headers
+        )
+        assert delete_response.status_code == 200
+
+        balance_after_delete = await async_client.get(
+            f"/api/v1/finance/users/{created_user['id']}/balance", headers=auth_headers
+        )
+        assert balance_after_delete.status_code == 200
+        assert balance_after_delete.json()["balance"] == 0.0  # Personal balance back to 0
+
+        remaining = await async_client.get(
+            f"/api/v1/finance/users/{created_user['id']}/transactions", headers=auth_headers
+        )
+        assert remaining.status_code == 200
+        assert len(remaining.json()) == 2  # 2 CC transactions remain
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_delete_cost_center_transaction_rebuilds_remaining_ledger(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        db_session,
+    ):
+        await self._enable_basic_user_creation(db_session)
+        created_user = await self._create_user_via_api(async_client, auth_headers, "erin")
+
+        shared_center = await db_session.scalar(
+            select(CostCenter).where(CostCenter.owner_user_id == created_user["id"], CostCenter.is_private.is_(True))
+        )
+        assert shared_center is not None
+
+        first_deposit = await async_client.post(
+            f"/api/v1/finance/users/{created_user['id']}/deposit",
+            json={"amount": 25.0, "description": "CC top-up", "cost_center_id": shared_center.id},
+            headers=auth_headers,
+        )
+        assert first_deposit.status_code == 200
+
+        cc_withdraw = await async_client.post(
+            f"/api/v1/finance/users/{created_user['id']}/withdraw",
+            json={"amount": 5.0, "description": "CC usage", "cost_center_id": shared_center.id},
+            headers=auth_headers,
+        )
+        assert cc_withdraw.status_code == 200
+
+        personal_deposit = await async_client.post(
+            f"/api/v1/finance/users/{created_user['id']}/deposit",
+            json={"amount": 12.0, "description": "Personal top-up", "cost_center_id": None},
+            headers=auth_headers,
+        )
+        assert personal_deposit.status_code == 200
+
+        delete_response = await async_client.delete(
+            f"/api/v1/finance/transactions/{first_deposit.json()['transaction']['id']}",
+            headers=auth_headers,
+        )
+        assert delete_response.status_code == 200
+
+        transactions_response = await async_client.get(
+            f"/api/v1/finance/users/{created_user['id']}/transactions", headers=auth_headers
+        )
+        assert transactions_response.status_code == 200
+        transactions = transactions_response.json()
+        assert len(transactions) == 2
+
+        cc_transaction = next(tx for tx in transactions if tx["cost_center_id"] == shared_center.id)
+        assert cc_transaction["amount"] == -5.0
+        assert cc_transaction["balance_after"] == -5.0
+
+        balance_response = await async_client.get(
+            f"/api/v1/finance/users/{created_user['id']}/balance", headers=auth_headers
+        )
+        assert balance_response.status_code == 200
+        assert balance_response.json()["balance"] == 12.0
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_delete_print_charge_stays_deleted_after_recalculate(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        db_session,
+    ):
+        await self._enable_basic_user_creation(db_session)
+        created_user = await self._create_user_via_api(async_client, auth_headers, "frank")
+        user = await db_session.scalar(select(User).where(User.id == created_user["id"]))
+        assert user is not None
+
+        archive = PrintArchive(
+            printer_id=None,
+            filename="print.gcode",
+            file_path="archives/test/print.gcode",
+            file_size=10,
+            content_hash="hash-print",
+            status="completed",
+            cost=4.0,
+            created_by_id=user.id,
+        )
+        db_session.add(archive)
+        await db_session.flush()
+
+        tx = WalletTransaction(
+            user_id=user.id,
+            transaction_type="print_charge",
+            amount=-4.0,
+            balance_after=-4.0,
+            description="Print charge: print.gcode",
+            created_by_user_id=None,
+            print_archive_id=archive.id,
+        )
+        db_session.add(tx)
+        await db_session.commit()
+
+        tx_rows_before = (
+            (await db_session.execute(select(WalletTransaction).where(WalletTransaction.user_id == user.id)))
+            .scalars()
+            .all()
+        )
+        assert len(tx_rows_before) == 1
+
+        delete_response = await async_client.delete(
+            f"/api/v1/finance/transactions/{tx_rows_before[0].id}", headers=auth_headers
+        )
+        assert delete_response.status_code == 200
+
+        tx_rows_after = (
+            (await db_session.execute(select(WalletTransaction).where(WalletTransaction.user_id == user.id)))
+            .scalars()
+            .all()
+        )
+        assert tx_rows_after == []
+
+    async def test_edit_transaction_updates_ledger(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        db_session,
+    ):
+        """Test that editing a transaction (user, cost_center, amount, description) rebuilds ledger."""
+        await self._enable_basic_user_creation(db_session)
+        user1 = await self._create_user_via_api(async_client, auth_headers, "user1")
+        user2 = await self._create_user_via_api(async_client, auth_headers, "user2")
+
+        # Create a cost center
+        cc_response = await async_client.post(
+            "/api/v1/finance/cost-centers",
+            json={"name": "Test Center", "is_active": True},
+            headers=auth_headers,
+        )
+        assert cc_response.status_code == 200
+        cost_center = cc_response.json()
+
+        # Get user records from DB
+        user1_db = await db_session.scalar(select(User).where(User.id == user1["id"]))
+        user2_db = await db_session.scalar(select(User).where(User.id == user2["id"]))
+
+        # Create a personal transaction for user1
+        tx_response = await async_client.post(
+            f"/api/v1/finance/users/{user1_db.id}/deposit",
+            json={"amount": 50.0, "description": "Initial deposit"},
+            headers=auth_headers,
+        )
+        assert tx_response.status_code == 200
+        tx_data = tx_response.json()
+        tx_id = tx_data["transaction"]["id"]
+
+        # Get the original transaction
+        original_tx = await db_session.scalar(select(WalletTransaction).where(WalletTransaction.id == tx_id))
+        assert original_tx.user_id == user1_db.id
+        assert original_tx.cost_center_id is None
+        assert original_tx.amount == 50.0
+        assert original_tx.balance_after == 50.0
+
+        # Edit the transaction: change user, add cost center, change amount
+        edit_response = await async_client.patch(
+            f"/api/v1/finance/transactions/{tx_id}",
+            json={
+                "user_id": user2_db.id,
+                "cost_center_id": cost_center["id"],
+                "amount": 75.0,
+                "description": "Updated deposit (Admin edit)",
+            },
+            headers=auth_headers,
+        )
+        assert edit_response.status_code == 200
+        edited_tx_data = edit_response.json()
+
+        # Verify transaction was updated
+        assert edited_tx_data["user_id"] == user2_db.id
+        assert edited_tx_data["cost_center_id"] == cost_center["id"]
+        assert edited_tx_data["amount"] == 75.0
+        # Description should have "(Admin edit)" appended
+        assert "(Admin edit)" in edited_tx_data["description"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_create_manual_print_and_recalculates_ledger(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        db_session,
+    ):
+        """Posting a manual print (manual_adjustment) creates a transaction and rebuilds ledger."""
+        await self._enable_basic_user_creation(db_session)
+        created_user = await self._create_user_via_api(async_client, auth_headers, "gina")
+
+        # Get user DB record
+        user = await db_session.scalar(select(User).where(User.id == created_user["id"]))
+        assert user is not None
+
+        # Private cost center for user
+        private_cc = await db_session.scalar(
+            select(CostCenter).where(CostCenter.owner_user_id == created_user["id"], CostCenter.is_private.is_(True))
+        )
+        assert private_cc is not None
+
+        # Post manual print affecting the cost center
+        payload = {
+            "user_id": user.id,
+            "cost_center_id": private_cc.id,
+            "amount": -4.0,
+            "description": "Manual adjustment for a print",
+            "created_at": "2026-05-12T12:00:00Z",
+        }
+
+        response = await async_client.post("/api/v1/finance/transactions/manual", json=payload, headers=auth_headers)
+        assert response.status_code == 200
+        resp_json = response.json()
+        assert "transaction" in resp_json or "id" in resp_json
+
+        # Response contains the created transaction details
+        assert resp_json["transaction_type"] == "manual_adjustment"
+        assert resp_json["amount"] == -4.0
+        assert resp_json["cost_center_id"] == private_cc.id
+
+        # The response includes the computed running balance for the transaction
+        assert resp_json.get("balance_after") == -4.0
+
+
+class TestPartialPrintChargesIntegration:
+    """Integration tests for partial print charge calculation."""
+
+    @pytest.fixture
+    async def admin_user(self, db_session):
+        user = User(
+            username="partial-admin",
+            email="partial-admin@example.com",
+            password_hash=get_password_hash("AdminPass1!"),
+            role="admin",
+            is_active=True,
+        )
+        db_session.add(user)
+        await db_session.commit()
+        await db_session.refresh(user)
+        return user
+
+    @pytest.fixture
+    async def auth_headers(self, async_client: AsyncClient, db_session, admin_user):
+        db_session.add(Settings(key="auth_enabled", value="true"))
+        db_session.add(Settings(key="advanced_auth_enabled", value="false"))
+        # Ensure billing is enabled for these partial-charge integration tests
+        existing = await db_session.scalar(select(Settings).where(Settings.key == "billing_enabled"))
+        if existing is None:
+            db_session.add(Settings(key="billing_enabled", value="true"))
+        else:
+            existing.value = "true"
+        await db_session.commit()
+
+        response = await async_client.post(
+            "/api/v1/auth/login",
+            json={"username": admin_user.username, "password": "AdminPass1!"},
+        )
+        assert response.status_code == 200
+        return {"Authorization": f"Bearer {response.json()['access_token']}"}
+
+    async def _create_user_via_api(self, async_client: AsyncClient, auth_headers: dict[str, str], username: str):
+        response = await async_client.post(
+            "/api/v1/users",
+            json={
+                "username": username,
+                "password": "Regularpass1!",
+                "email": f"{username}@example.com",
+                "role": "user",
+            },
+            headers=auth_headers,
+        )
+        assert response.status_code == 201
+        return response.json()
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_aborted_print_charges_proportionally_via_recalculate_endpoint(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        db_session,
+    ):
+        """Verify aborted prints are included in recalculate and charged proportionally."""
+        created_user = await self._create_user_via_api(async_client, auth_headers, "frank")
+        user = await db_session.scalar(select(User).where(User.id == created_user["id"]))
+        assert user is not None
+
+        # Wallet is already created by ensure_user_finance_defaults during user creation
+
+        # Archive: completed print (100% charge)
+        completed = PrintArchive(
+            printer_id=None,
+            filename="completed.3mf",
+            file_path="archives/test/completed.3mf",
+            file_size=100,
+            content_hash="partial-complete",
+            status="completed",
+            cost=10.0,
+            created_by_id=user.id,
+        )
+
+        # Archive: aborted print (50% filament used = 50% charge)
+        aborted = PrintArchive(
+            printer_id=None,
+            filename="aborted.3mf",
+            file_path="archives/test/aborted.3mf",
+            file_size=100,
+            content_hash="partial-aborted",
+            status="aborted",
+            cost=8.0,
+            filament_used_grams=50.0,
+            extra_data={"filament_grams_total": 100.0},
+            created_by_id=user.id,
+        )
+
+        # Archive: failed print (0% filament used = no charge)
+        failed = PrintArchive(
+            printer_id=None,
+            filename="failed.3mf",
+            file_path="archives/test/failed.3mf",
+            file_size=100,
+            content_hash="partial-failed",
+            status="failed",
+            cost=5.0,
+            filament_used_grams=0.0,
+            created_by_id=user.id,
+        )
+
+        db_session.add_all([completed, aborted, failed])
+        await db_session.commit()
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_partial_charges_appear_in_transaction_ledger(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        db_session,
+    ):
+        """Verify transaction descriptions indicate partial charges."""
+        created_user = await self._create_user_via_api(async_client, auth_headers, "grace")
+        user = await db_session.scalar(select(User).where(User.id == created_user["id"]))
+        assert user is not None
+
+        # Wallet is already created by ensure_user_finance_defaults during user creation
+
+        cancelled = PrintArchive(
+            printer_id=None,
+            filename="cancelled.3mf",
+            file_path="archives/test/cancelled.3mf",
+            file_size=100,
+            content_hash="partial-cancel",
+            status="cancelled",
+            cost=12.0,
+            filament_used_grams=25.0,
+            extra_data={"filament_grams_total": 100.0},
+            print_name="Partially Cancelled Print",
+            created_by_id=user.id,
+        )
+        db_session.add(cancelled)
+        await db_session.commit()
+
+        from backend.app.services.finance_billing import apply_print_charge_for_archive
+
+        changed = await apply_print_charge_for_archive(db_session, cancelled.id)
+        assert changed is True
+        await db_session.commit()
+
+        tx_response = await async_client.get(
+            f"/api/v1/finance/users/{created_user['id']}/transactions", headers=auth_headers
+        )
+        assert tx_response.status_code == 200
+        transactions = tx_response.json()
+        assert len(transactions) == 1
+
+        tx = transactions[0]
+        assert tx["transaction_type"] == "print_charge"
+        assert tx["amount"] == -3.0  # 25% of 12.0
+        assert "cancelled" in tx["description"].lower()
+        assert "25.0g/100.0" in tx["description"]  # filament amounts in description
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_partial_charges_with_cost_center_override(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        db_session,
+    ):
+        """Verify partial charges respect cost_center_id when present."""
+        created_user = await self._create_user_via_api(async_client, auth_headers, "henry")
+        user = await db_session.scalar(select(User).where(User.id == created_user["id"]))
+        assert user is not None
+
+        # Create cost centers
+        default_cc = CostCenter(name="Default CC", owner_user_id=user.id, is_active=True, is_private=False)
+        lab_cc = CostCenter(name="Lab CC", owner_user_id=user.id, is_active=True, is_private=False)
+        db_session.add_all([default_cc, lab_cc])
+        await db_session.flush()
+
+        # Wallet is already created by ensure_user_finance_defaults during user creation
+
+        # Archive assigned to default_cc
+        aborted = PrintArchive(
+            printer_id=None,
+            filename="aborted_cc.3mf",
+            file_path="archives/test/aborted_cc.3mf",
+            file_size=100,
+            content_hash="partial-cc",
+            status="aborted",
+            cost=6.0,
+            filament_used_grams=30.0,
+            extra_data={"filament_grams_total": 100.0},
+            cost_center_id=default_cc.id,
+            created_by_id=user.id,
+        )
+        db_session.add(aborted)
+        await db_session.commit()
+
+        # Manually apply charge with override
+        from backend.app.services.finance_billing import apply_print_charge_for_archive
+
+        changed = await apply_print_charge_for_archive(
+            db_session,
+            aborted.id,
+            cost_center_id=lab_cc.id,
+        )
+        await db_session.commit()
+
+        assert changed is True
+
+        tx_response = await async_client.get(
+            f"/api/v1/finance/users/{created_user['id']}/transactions", headers=auth_headers
+        )
+        assert tx_response.status_code == 200
+        transactions = tx_response.json()
+        assert len(transactions) == 1
+
+        tx = transactions[0]
+        assert tx["cost_center_id"] == lab_cc.id  # Overridden to lab_cc
+        assert tx["amount"] == pytest.approx(-1.8, abs=0.01)  # 30% of 6.0
+
+
+class TestFinanceUserDefaults:
+    """Tests for user creation and finance defaults initialization."""
+
+    @pytest.fixture
+    async def admin_user(self, db_session):
+        user = User(
+            username="billing-admin",
+            email="billing-admin@example.com",
+            password_hash=get_password_hash("AdminPass1!"),
+            role="admin",
+            is_active=True,
+        )
+        db_session.add(user)
+        await db_session.commit()
+        await db_session.refresh(user)
+        return user
+
+    @pytest.fixture
+    async def auth_headers(self, async_client: AsyncClient, db_session, admin_user):
+        db_session.add(Settings(key="auth_enabled", value="true"))
+        db_session.add(Settings(key="advanced_auth_enabled", value="false"))
+        await db_session.commit()
+
+        response = await async_client.post(
+            "/api/v1/auth/login",
+            json={"username": admin_user.username, "password": "AdminPass1!"},
+        )
+        assert response.status_code == 200
+        return {"Authorization": f"Bearer {response.json()['access_token']}"}
+
+    async def _create_user_via_api(self, async_client: AsyncClient, auth_headers: dict[str, str], username: str):
+        response = await async_client.post(
+            "/api/v1/users",
+            json={
+                "username": username,
+                "password": "Regularpass1!",
+                "email": f"{username}@example.com",
+                "role": "user",
+            },
+            headers=auth_headers,
+        )
+        assert response.status_code == 201
+        return response.json()
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_create_user_initializes_wallet_and_private_cost_center(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        db_session,
+    ):
+        """Verify user creation initializes wallet, private cost center, and membership."""
+        result = await async_client.post(
+            "/api/v1/users",
+            json={
+                "username": "alice",
+                "password": "Regularpass1!",
+                "email": "alice@example.com",
+                "role": "user",
+            },
+            headers=auth_headers,
+        )
+
+        assert result.status_code == 201
+        created = result.json()
+        assert created["username"] == "alice"
+
+        user = await db_session.scalar(select(User).where(User.username == "alice"))
+        assert user is not None
+
+        from backend.app.models.finance import CostCenterMember
+
+        wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
+        assert wallet is not None
+        assert wallet.balance == 0.0
+
+        private_center = await db_session.scalar(
+            select(CostCenter).where(CostCenter.owner_user_id == user.id, CostCenter.is_private.is_(True))
+        )
+        assert private_center is not None
+        assert private_center.name == "alice"
+
+        membership = await db_session.scalar(
+            select(CostCenterMember).where(
+                CostCenterMember.cost_center_id == private_center.id,
+                CostCenterMember.user_id == user.id,
+            )
+        )
+        assert membership is not None
+        assert membership.can_print is True
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_update_user_keeps_private_cost_center_in_sync(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        db_session,
+    ):
+        """Verify user updates keep private cost center name in sync."""
+        created = await self._create_user_via_api(async_client, auth_headers, "bob")
+
+        response = await async_client.patch(
+            f"/api/v1/users/{created['id']}",
+            json={"username": "bobby"},
+            headers=auth_headers,
+        )
+
+        assert response.status_code == 200
+        assert response.json()["username"] == "bobby"
+
+        user = await db_session.scalar(select(User).where(User.id == created["id"]))
+        assert user is not None
+
+        private_centers = (
+            (
+                await db_session.execute(
+                    select(CostCenter).where(CostCenter.owner_user_id == user.id, CostCenter.is_private.is_(True))
+                )
+            )
+            .scalars()
+            .all()
+        )
+        assert len(private_centers) == 1
+        assert private_centers[0].name == "bobby"
+
+        wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
+        assert wallet is not None

backend/tests/integration/test_print_queue_api.py

@@ -2,6 +2,19 @@
 
 import pytest
 from httpx import AsyncClient
+from sqlalchemy import select
+
+from backend.app.models.finance import CostCenter
+from backend.app.models.settings import Settings
+
+
+async def enable_billing(db_session):
+    setting = await db_session.scalar(select(Settings).where(Settings.key == "billing_enabled"))
+    if setting is None:
+        db_session.add(Settings(key="billing_enabled", value="true"))
+    else:
+        setting.value = "true"
+    await db_session.commit()
 
 
 class TestPrintQueueAPI:
@@ -125,6 +138,149 @@ class TestPrintQueueAPI:
         assert result["status"] == "pending"
         assert result["manual_start"] is False
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_add_to_queue_with_cost_center_id(
+        self, async_client: AsyncClient, printer_factory, archive_factory, db_session
+    ):
+        """Verify item can be added to queue with cost_center_id."""
+        await enable_billing(db_session)
+        printer = await printer_factory()
+        archive = await archive_factory()
+        cost_center = CostCenter(name="Queue CC", is_active=True, is_private=False)
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+
+        response = await async_client.post(
+            "/api/v1/queue/",
+            json={
+                "printer_id": printer.id,
+                "archive_id": archive.id,
+                "cost_center_id": cost_center.id,
+                "estimated_cost": 1.25,
+            },
+        )
+        assert response.status_code == 200
+        result = response.json()
+        assert result["cost_center_id"] == cost_center.id
+        assert result["estimated_cost"] == 1.25
+
+        from backend.app.models.print_queue import PrintQueueItem
+
+        row = await db_session.scalar(select(PrintQueueItem).where(PrintQueueItem.id == result["id"]))
+        assert row is not None
+        assert row.cost_center_id == cost_center.id
+        assert row.estimated_cost == 1.25
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_add_to_queue_with_cost_center_requires_estimated_cost(
+        self, async_client: AsyncClient, printer_factory, archive_factory, db_session
+    ):
+        """Cost-center queue items require an estimated cost for budget checks."""
+        await enable_billing(db_session)
+        printer = await printer_factory()
+        archive = await archive_factory()
+        cost_center = CostCenter(name="Budget CC", is_active=True, is_private=False, monthly_budget=10.0)
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+
+        response = await async_client.post(
+            "/api/v1/queue/",
+            json={
+                "printer_id": printer.id,
+                "archive_id": archive.id,
+                "cost_center_id": cost_center.id,
+            },
+        )
+
+        assert response.status_code == 400
+        assert "Estimated cost is required" in response.json()["detail"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_add_to_queue_rejects_when_estimated_cost_exceeds_budget(
+        self, async_client: AsyncClient, printer_factory, archive_factory, db_session
+    ):
+        """Queue creation is rejected if the estimated cost exceeds remaining budget."""
+        await enable_billing(db_session)
+        printer = await printer_factory()
+        archive = await archive_factory()
+        cost_center = CostCenter(name="Tiny Budget CC", is_active=True, is_private=False, monthly_budget=1.0)
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+
+        response = await async_client.post(
+            "/api/v1/queue/",
+            json={
+                "printer_id": printer.id,
+                "archive_id": archive.id,
+                "cost_center_id": cost_center.id,
+                "estimated_cost": 2.0,
+            },
+        )
+
+        assert response.status_code == 400
+        assert "exceeds available cost center budget" in response.json()["detail"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_add_to_queue_requires_cost_center_when_billing_enabled(
+        self, async_client: AsyncClient, printer_factory, archive_factory, db_session
+    ):
+        """Billing enforcement rejects queue jobs that omit cost center."""
+        await enable_billing(db_session)
+        printer = await printer_factory()
+        archive = await archive_factory()
+
+        response = await async_client.post(
+            "/api/v1/queue/",
+            json={
+                "printer_id": printer.id,
+                "archive_id": archive.id,
+            },
+        )
+
+        assert response.status_code == 400
+        assert "Cost center is required" in response.json()["detail"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_add_to_queue_counts_pending_queue_reservations_against_budget(
+        self, async_client: AsyncClient, printer_factory, archive_factory, queue_item_factory, db_session
+    ):
+        """Open queue items reserve budget until they leave pending/printing states."""
+        await enable_billing(db_session)
+        printer = await printer_factory()
+        archive = await archive_factory()
+        cost_center = CostCenter(name="Reserved Budget CC", is_active=True, is_private=False, monthly_budget=10.0)
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+        await queue_item_factory(
+            printer_id=printer.id,
+            archive_id=archive.id,
+            cost_center_id=cost_center.id,
+            estimated_cost=8.0,
+            status="pending",
+        )
+
+        response = await async_client.post(
+            "/api/v1/queue/",
+            json={
+                "printer_id": printer.id,
+                "archive_id": archive.id,
+                "cost_center_id": cost_center.id,
+                "estimated_cost": 3.0,
+            },
+        )
+
+        assert response.status_code == 400
+        assert "exceeds available cost center budget" in response.json()["detail"]
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_add_to_queue_with_manual_start(
@@ -181,6 +337,30 @@ class TestPrintQueueAPI:
         result = response.json()
         assert result["skip_filament_check"] is False
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_update_queue_item_cost_center_id(
+        self, async_client: AsyncClient, printer_factory, archive_factory, queue_item_factory, db_session
+    ):
+        """Verify a pending queue item can be updated with cost_center_id."""
+        await enable_billing(db_session)
+        printer = await printer_factory()
+        archive = await archive_factory()
+        item = await queue_item_factory(printer_id=printer.id, archive_id=archive.id)
+        cost_center = CostCenter(name="Update CC", is_active=True, is_private=False)
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+
+        response = await async_client.patch(
+            f"/api/v1/queue/{item.id}",
+            json={"cost_center_id": cost_center.id, "estimated_cost": 1.25},
+        )
+
+        assert response.status_code == 200
+        assert response.json()["cost_center_id"] == cost_center.id
+        assert response.json()["estimated_cost"] == 1.25
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_add_to_queue_with_project_id(
@@ -378,6 +558,54 @@ class TestPrintQueueAPI:
         assert response.status_code == 200
         assert response.json()["message"] == "Queue item deleted"
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_delete_queue_item_releases_reserved_budget(
+        self, async_client: AsyncClient, printer_factory, archive_factory, queue_item_factory, db_session
+    ):
+        """Deleting a pending cost-center queue item releases its reserved budget."""
+        await enable_billing(db_session)
+        printer = await printer_factory()
+        archive = await archive_factory()
+        cost_center = CostCenter(
+            name="Delete Releases Budget CC", is_active=True, is_private=False, monthly_budget=10.0
+        )
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+        item = await queue_item_factory(
+            printer_id=printer.id,
+            archive_id=archive.id,
+            cost_center_id=cost_center.id,
+            estimated_cost=8.0,
+            status="pending",
+        )
+
+        blocked = await async_client.post(
+            "/api/v1/queue/",
+            json={
+                "printer_id": printer.id,
+                "archive_id": archive.id,
+                "cost_center_id": cost_center.id,
+                "estimated_cost": 3.0,
+            },
+        )
+        assert blocked.status_code == 400
+
+        deleted = await async_client.delete(f"/api/v1/queue/{item.id}")
+        assert deleted.status_code == 200
+
+        allowed = await async_client.post(
+            "/api/v1/queue/",
+            json={
+                "printer_id": printer.id,
+                "archive_id": archive.id,
+                "cost_center_id": cost_center.id,
+                "estimated_cost": 3.0,
+            },
+        )
+        assert allowed.status_code == 200
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_delete_queue_item_not_found(self, async_client: AsyncClient):
@@ -504,6 +732,24 @@ class TestQueueStartEndpoint:
         result = response.json()
         assert result["manual_start"] is False
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_start_queue_item_with_cost_center_requires_estimated_cost(
+        self, async_client: AsyncClient, queue_item_factory, db_session
+    ):
+        """Starting a cost-center queue item requires a stored estimate."""
+        await enable_billing(db_session)
+        cost_center = CostCenter(name="Start Budget CC", is_active=True, is_private=False, monthly_budget=10.0)
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+        item = await queue_item_factory(manual_start=True, cost_center_id=cost_center.id, estimated_cost=None)
+
+        response = await async_client.post(f"/api/v1/queue/{item.id}/start")
+
+        assert response.status_code == 400
+        assert "Estimated cost is required" in response.json()["detail"]
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_start_queue_item_not_found(self, async_client: AsyncClient):
@@ -1188,6 +1434,57 @@ class TestBulkUpdateEndpoint:
         assert response.status_code == 400
         assert "printer not found" in response.json()["detail"].lower()
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_bulk_update_cost_center_requires_estimated_cost(
+        self, async_client: AsyncClient, queue_item_factory, db_session
+    ):
+        """Bulk assigning a cost center requires an estimate, same as single-item updates."""
+        await enable_billing(db_session)
+        item = await queue_item_factory()
+        cost_center = CostCenter(name="Bulk Budget CC", is_active=True, is_private=False, monthly_budget=10.0)
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+
+        response = await async_client.patch(
+            "/api/v1/queue/bulk",
+            json={"item_ids": [item.id], "cost_center_id": cost_center.id},
+        )
+
+        assert response.status_code == 400
+        assert "Estimated cost is required" in response.json()["detail"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_bulk_update_cost_center_counts_pending_reservations(
+        self, async_client: AsyncClient, queue_item_factory, db_session
+    ):
+        """Bulk updates cannot reserve more than the available cost-center budget."""
+        await enable_billing(db_session)
+        item = await queue_item_factory()
+        existing = await queue_item_factory()
+        cost_center = CostCenter(name="Bulk Reserved CC", is_active=True, is_private=False, monthly_budget=10.0)
+        db_session.add(cost_center)
+        await db_session.commit()
+        await db_session.refresh(cost_center)
+
+        existing.cost_center_id = cost_center.id
+        existing.estimated_cost = 8.0
+        await db_session.commit()
+
+        response = await async_client.patch(
+            "/api/v1/queue/bulk",
+            json={"item_ids": [item.id], "cost_center_id": cost_center.id, "estimated_cost": 3.0},
+        )
+
+        assert response.status_code == 400
+        assert "exceeds available cost center budget" in response.json()["detail"]
+
+        await db_session.refresh(item)
+        assert item.cost_center_id is None
+        assert item.estimated_cost is None
+
 
 class TestTargetLocationFeature:
     """Tests for queue items with target_location (Issue #220)."""

backend/tests/unit/services/test_finance_service_billing.py

@@ -0,0 +1,481 @@
+"""Unit tests for billing charges applied to print archives."""
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models.archive import PrintArchive
+from backend.app.models.finance import BudgetReservation, CostCenter, UserWallet, WalletTransaction
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+from backend.app.services.finance_billing import apply_print_charge_for_archive
+
+
+async def enable_billing(db_session):
+    setting = await db_session.scalar(select(Settings).where(Settings.key == "billing_enabled"))
+    if setting is None:
+        db_session.add(Settings(key="billing_enabled", value="true"))
+    else:
+        setting.value = "true"
+    await db_session.commit()
+
+
+class TestFinanceBilling:
+    @pytest.mark.asyncio
+    async def test_apply_print_charge_uses_print_run_id_and_cost_center_override(self, db_session):
+        await enable_billing(db_session)
+        user = User(username="printer", role="user", is_active=True)
+        archive_cost_center = CostCenter(name="Archive CC", is_active=True, is_private=False)
+        override_cost_center = CostCenter(name="Override CC", is_active=True, is_private=False)
+        db_session.add_all([user, archive_cost_center, override_cost_center])
+        await db_session.commit()
+        await db_session.refresh(user)
+        await db_session.refresh(archive_cost_center)
+        await db_session.refresh(override_cost_center)
+
+        archive = PrintArchive(
+            printer_id=None,
+            filename="test.3mf",
+            file_path="archives/test/test.3mf",
+            file_size=123,
+            content_hash="hash-1",
+            status="completed",
+            cost=7.5,
+            created_by_id=user.id,
+            cost_center_id=archive_cost_center.id,
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+
+        changed = await apply_print_charge_for_archive(
+            db_session,
+            archive.id,
+            cost_center_id=override_cost_center.id,
+            print_run_id="run-1",
+        )
+        await db_session.commit()
+
+        assert changed is True
+        assert archive.cost_center_id == archive_cost_center.id
+
+        wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
+        assert wallet is not None
+        assert wallet.balance == -7.5
+
+        tx = await db_session.scalar(select(WalletTransaction).where(WalletTransaction.print_run_id == "run-1"))
+        assert tx is not None
+        assert tx.cost_center_id == override_cost_center.id
+        assert tx.print_archive_id == archive.id
+
+        duplicate = await apply_print_charge_for_archive(
+            db_session,
+            archive.id,
+            cost_center_id=override_cost_center.id,
+            print_run_id="run-1",
+        )
+        assert duplicate is False
+
+        second_run = await apply_print_charge_for_archive(
+            db_session,
+            archive.id,
+            cost_center_id=override_cost_center.id,
+            print_run_id="run-2",
+        )
+        await db_session.commit()
+
+        assert second_run is True
+        wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
+        assert wallet is not None
+        assert wallet.balance == -15.0
+
+        rows = (
+            (await db_session.execute(select(WalletTransaction).where(WalletTransaction.user_id == user.id)))
+            .scalars()
+            .all()
+        )
+        assert len(rows) == 2
+        assert {row.print_run_id for row in rows} == {"run-1", "run-2"}
+
+    @pytest.mark.asyncio
+    async def test_apply_print_charge_consumes_matching_budget_reservation(self, db_session):
+        await enable_billing(db_session)
+        user = User(username="reserved", role="user", is_active=True)
+        cost_center = CostCenter(name="Reserved CC", is_active=True, is_private=False)
+        db_session.add_all([user, cost_center])
+        await db_session.commit()
+        await db_session.refresh(user)
+        await db_session.refresh(cost_center)
+
+        archive = PrintArchive(
+            printer_id=None,
+            filename="reserved.3mf",
+            file_path="archives/test/reserved.3mf",
+            file_size=123,
+            content_hash="hash-reserved",
+            status="completed",
+            cost=4.0,
+            created_by_id=user.id,
+            cost_center_id=cost_center.id,
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+
+        reservation = BudgetReservation(
+            cost_center_id=cost_center.id,
+            amount=4.0,
+            status="active",
+            source_type="background_dispatch",
+            source_id=42,
+            print_archive_id=archive.id,
+        )
+        db_session.add(reservation)
+        await db_session.commit()
+        await db_session.refresh(reservation)
+
+        changed = await apply_print_charge_for_archive(db_session, archive.id, print_run_id="run-reserved")
+        await db_session.commit()
+
+        assert changed is True
+        await db_session.refresh(reservation)
+        assert reservation.status == "consumed"
+        assert reservation.released_at is not None
+
+    @pytest.mark.asyncio
+    async def test_apply_print_charge_rejects_ineligible_archive(self, db_session):
+        await enable_billing(db_session)
+        user = User(username="skipped", role="user", is_active=True)
+        db_session.add(user)
+        await db_session.commit()
+        await db_session.refresh(user)
+
+        # Reject print with unknown status
+        archive = PrintArchive(
+            printer_id=None,
+            filename="unknown.3mf",
+            file_path="archives/test/unknown.3mf",
+            file_size=123,
+            content_hash="hash-2",
+            status="unknown",
+            cost=1.0,
+            created_by_id=user.id,
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+
+        changed = await apply_print_charge_for_archive(db_session, archive.id, print_run_id="run-unknown")
+        assert changed is False
+
+    @pytest.mark.asyncio
+    async def test_apply_print_charge_skips_when_billing_disabled(self, db_session):
+        user = User(username="billing_disabled", role="user", is_active=True)
+        cost_center = CostCenter(name="Disabled Billing CC", is_active=True, is_private=False)
+        db_session.add_all([user, cost_center])
+        await db_session.commit()
+        await db_session.refresh(user)
+        await db_session.refresh(cost_center)
+
+        archive = PrintArchive(
+            printer_id=None,
+            filename="billing-disabled.3mf",
+            file_path="archives/test/billing-disabled.3mf",
+            file_size=123,
+            content_hash="hash-disabled-billing",
+            status="completed",
+            cost=7.5,
+            created_by_id=user.id,
+            cost_center_id=cost_center.id,
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+        reservation = BudgetReservation(
+            cost_center_id=cost_center.id,
+            amount=7.5,
+            status="active",
+            source_type="background_dispatch",
+            source_id=123,
+            print_archive_id=archive.id,
+        )
+        db_session.add(reservation)
+        await db_session.commit()
+        await db_session.refresh(reservation)
+
+        changed = await apply_print_charge_for_archive(db_session, archive.id, print_run_id="run-disabled")
+        await db_session.commit()
+
+        assert changed is False
+        wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
+        tx = await db_session.scalar(select(WalletTransaction).where(WalletTransaction.print_run_id == "run-disabled"))
+        assert wallet is None
+        assert tx is None
+        await db_session.refresh(reservation)
+        assert reservation.status == "released"
+        assert reservation.released_at is not None
+
+
+class TestPartialPrintCharges:
+    """Tests for proportional charge calculation on aborted/failed/cancelled prints."""
+
+    @pytest.mark.asyncio
+    async def test_partial_print_with_missing_planned_filament_is_skipped(self, db_session):
+        await enable_billing(db_session)
+        user = User(username="missing_plan", role="user", is_active=True)
+        cost_center = CostCenter(name="Missing Plan CC", is_active=True, is_private=False)
+        db_session.add_all([user, cost_center])
+        await db_session.commit()
+        await db_session.refresh(user)
+        await db_session.refresh(cost_center)
+
+        archive = PrintArchive(
+            printer_id=None,
+            filename="missing-plan.3mf",
+            file_path="archives/test/missing-plan.3mf",
+            file_size=100,
+            content_hash="missing-plan-hash",
+            status="aborted",
+            cost=12.0,
+            filament_used_grams=80.0,
+            created_by_id=user.id,
+            cost_center_id=cost_center.id,
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+
+        changed = await apply_print_charge_for_archive(db_session, archive.id)
+        await db_session.commit()
+
+        assert changed is False
+        wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
+        assert wallet is None
+
+    @pytest.mark.asyncio
+    async def test_invalid_transaction_type_is_rejected(self, db_session):
+        user = User(username="invalid_tx", role="user", is_active=True)
+        db_session.add(user)
+        await db_session.commit()
+        await db_session.refresh(user)
+
+        with pytest.raises(ValueError, match="Invalid transaction type"):
+            WalletTransaction(
+                user_id=user.id,
+                transaction_type="not-a-real-type",
+                amount=1.0,
+            )
+
+    @pytest.mark.asyncio
+    async def test_aborted_print_with_partial_filament_charges_proportionally(self, db_session):
+        """Verify aborted print charges proportionally based on filament used."""
+        await enable_billing(db_session)
+        user = User(username="abort_test", role="user", is_active=True)
+        cost_center = CostCenter(name="Abort CC", is_active=True, is_private=False)
+        db_session.add_all([user, cost_center])
+        await db_session.commit()
+        await db_session.refresh(user)
+        await db_session.refresh(cost_center)
+
+        # Archive with 100g planned, but only 50g used (50% filament)
+        archive = PrintArchive(
+            printer_id=None,
+            filename="abort.3mf",
+            file_path="archives/test/abort.3mf",
+            file_size=100,
+            content_hash="abort-hash",
+            status="aborted",
+            cost=10.0,  # Full cost would be 10.0
+            filament_used_grams=50.0,
+            extra_data={"filament_grams_total": 100.0},
+            created_by_id=user.id,
+            cost_center_id=cost_center.id,
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+
+        changed = await apply_print_charge_for_archive(db_session, archive.id)
+        await db_session.commit()
+
+        assert changed is True
+
+        wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
+        assert wallet is not None
+        assert wallet.balance == -5.0  # 50% of 10.0
+
+        tx = await db_session.scalar(
+            select(WalletTransaction)
+            .where(WalletTransaction.user_id == user.id)
+            .where(WalletTransaction.transaction_type == "print_charge")
+        )
+        assert tx is not None
+        assert tx.amount == -5.0
+        assert "aborted" in tx.description.lower()
+        assert "50.0" in tx.description  # filament used
+
+    @pytest.mark.asyncio
+    async def test_cancelled_print_with_no_filament_data_is_not_charged(self, db_session):
+        """Verify cancelled print with no filament data is skipped."""
+        await enable_billing(db_session)
+        user = User(username="cancel_no_data", role="user", is_active=True)
+        cost_center = CostCenter(name="Cancel No Data CC", is_active=True, is_private=False)
+        db_session.add_all([user, cost_center])
+        await db_session.commit()
+        await db_session.refresh(user)
+        await db_session.refresh(cost_center)
+
+        archive = PrintArchive(
+            printer_id=None,
+            filename="cancel.3mf",
+            file_path="archives/test/cancel.3mf",
+            file_size=100,
+            content_hash="cancel-hash",
+            status="cancelled",
+            cost=5.0,
+            filament_used_grams=None,  # No data
+            created_by_id=user.id,
+            cost_center_id=cost_center.id,
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+        reservation = BudgetReservation(
+            cost_center_id=cost_center.id,
+            amount=5.0,
+            status="active",
+            source_type="background_dispatch",
+            source_id=99,
+            print_archive_id=archive.id,
+        )
+        db_session.add(reservation)
+        await db_session.commit()
+        await db_session.refresh(reservation)
+
+        changed = await apply_print_charge_for_archive(db_session, archive.id)
+        await db_session.commit()
+
+        assert changed is False
+        wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
+        assert wallet is None  # No wallet created
+        await db_session.refresh(reservation)
+        assert reservation.status == "released"
+        assert reservation.released_at is not None
+
+    @pytest.mark.asyncio
+    async def test_failed_print_with_minimal_filament_charges_small_amount(self, db_session):
+        """Verify failed print with minimal filament usage charges proportionally."""
+        await enable_billing(db_session)
+        user = User(username="fail_min", role="user", is_active=True)
+        cost_center = CostCenter(name="Fail Min CC", is_active=True, is_private=False)
+        db_session.add_all([user, cost_center])
+        await db_session.commit()
+        await db_session.refresh(user)
+        await db_session.refresh(cost_center)
+
+        # 5% filament used out of 100g planned
+        archive = PrintArchive(
+            printer_id=None,
+            filename="fail_min.3mf",
+            file_path="archives/test/fail_min.3mf",
+            file_size=100,
+            content_hash="fail-min-hash",
+            status="failed",
+            cost=20.0,
+            filament_used_grams=5.0,
+            extra_data={"filament_grams_total": 100.0},
+            failure_reason="Filament runout",
+            created_by_id=user.id,
+            cost_center_id=cost_center.id,
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+
+        changed = await apply_print_charge_for_archive(db_session, archive.id)
+        await db_session.commit()
+
+        assert changed is True
+
+        wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
+        assert wallet is not None
+        assert wallet.balance == pytest.approx(-1.0, abs=0.01)  # 5% of 20.0
+
+    @pytest.mark.asyncio
+    async def test_completed_print_still_charges_full_cost(self, db_session):
+        """Verify completed prints ignore filament ratio and charge full cost."""
+        await enable_billing(db_session)
+        user = User(username="completed_full", role="user", is_active=True)
+        cost_center = CostCenter(name="Completed Full CC", is_active=True, is_private=False)
+        db_session.add_all([user, cost_center])
+        await db_session.commit()
+        await db_session.refresh(user)
+        await db_session.refresh(cost_center)
+
+        archive = PrintArchive(
+            printer_id=None,
+            filename="complete.3mf",
+            file_path="archives/test/complete.3mf",
+            file_size=100,
+            content_hash="complete-hash",
+            status="completed",
+            cost=15.0,
+            filament_used_grams=100.0,
+            extra_data={"filament_grams_total": 100.0},
+            created_by_id=user.id,
+            cost_center_id=cost_center.id,
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+
+        changed = await apply_print_charge_for_archive(db_session, archive.id)
+        await db_session.commit()
+
+        assert changed is True
+
+        wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
+        assert wallet.balance == -15.0  # Full cost, not proportional
+
+    @pytest.mark.asyncio
+    async def test_partial_charge_with_cost_center_override(self, db_session):
+        """Verify partial charges respect cost_center_id override."""
+        await enable_billing(db_session)
+        user = User(username="partial_cc", role="user", is_active=True)
+        default_cc = CostCenter(name="Default", is_active=True, is_private=False)
+        override_cc = CostCenter(name="Override", is_active=True, is_private=False)
+        db_session.add_all([user, default_cc, override_cc])
+        await db_session.commit()
+        await db_session.refresh(user)
+        await db_session.refresh(default_cc)
+        await db_session.refresh(override_cc)
+
+        archive = PrintArchive(
+            printer_id=None,
+            filename="partial_cc.3mf",
+            file_path="archives/test/partial_cc.3mf",
+            file_size=100,
+            content_hash="partial-cc-hash",
+            status="aborted",
+            cost=8.0,
+            filament_used_grams=25.0,
+            extra_data={"filament_grams_total": 100.0},
+            cost_center_id=default_cc.id,
+            created_by_id=user.id,
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+
+        changed = await apply_print_charge_for_archive(db_session, archive.id, cost_center_id=override_cc.id)
+        await db_session.commit()
+
+        assert changed is True
+
+        tx = await db_session.scalar(
+            select(WalletTransaction)
+            .where(WalletTransaction.user_id == user.id)
+            .where(WalletTransaction.transaction_type == "print_charge")
+        )
+        assert tx is not None
+        assert tx.cost_center_id == override_cc.id
+        assert tx.amount == -2.0  # 25% of 8.0

backend/tests/unit/services/test_finance_service_defaults.py

@@ -0,0 +1,72 @@
+"""Unit tests for finance defaults applied during user creation/update."""
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models.finance import CostCenter, CostCenterMember, UserWallet
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+from backend.app.services.finance_defaults import ensure_user_finance_defaults
+
+
+class TestFinanceDefaults:
+    @pytest.mark.asyncio
+    async def test_creates_wallet_private_center_and_membership(self, db_session):
+        db_session.add(Settings(key="currency", value="USD"))
+
+        user = User(username="alice", role="user", is_active=True)
+        db_session.add(user)
+        await db_session.commit()
+        await db_session.refresh(user)
+
+        changed = await ensure_user_finance_defaults(db_session, user)
+        await db_session.commit()
+
+        assert changed is True
+
+        wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
+        assert wallet is not None
+        assert wallet.balance == 0.0
+        assert wallet.currency == "USD"
+
+        center = await db_session.scalar(
+            select(CostCenter).where(CostCenter.owner_user_id == user.id, CostCenter.is_private.is_(True))
+        )
+        assert center is not None
+        assert center.name == "alice"
+
+        membership = await db_session.scalar(
+            select(CostCenterMember).where(
+                CostCenterMember.cost_center_id == center.id,
+                CostCenterMember.user_id == user.id,
+            )
+        )
+        assert membership is not None
+        assert membership.can_print is True
+
+    @pytest.mark.asyncio
+    async def test_updates_private_center_name_and_is_idempotent(self, db_session):
+        user = User(username="bob", role="user", is_active=True)
+        db_session.add(user)
+        await db_session.commit()
+        await db_session.refresh(user)
+
+        initial_changed = await ensure_user_finance_defaults(db_session, user)
+        await db_session.commit()
+
+        assert initial_changed is True
+
+        user.username = "bobby"
+        renamed_changed = await ensure_user_finance_defaults(db_session, user)
+        await db_session.commit()
+
+        assert renamed_changed is True
+
+        center = await db_session.scalar(
+            select(CostCenter).where(CostCenter.owner_user_id == user.id, CostCenter.is_private.is_(True))
+        )
+        assert center is not None
+        assert center.name == "bobby"
+
+        idempotent_changed = await ensure_user_finance_defaults(db_session, user)
+        assert idempotent_changed is False

backend/tests/unit/test_phantom_print_hardening.py

@@ -9,6 +9,7 @@ NOT by calling the full on_print_start/on_print_complete callbacks
 """
 
 import logging
+from unittest.mock import patch
 
 import pytest
 from sqlalchemy import or_, select
@@ -346,3 +347,125 @@ class TestBusyPrinterSeedingFromPrintingItems:
         assert statuses == ["printing", "pending"]
 
         await engine.dispose()
+
+    @pytest.mark.asyncio
+    async def test_scheduler_budget_check_uses_queue_creator_membership(self):
+        """Scheduler must not bypass cost-center membership when auto-starting a queued job."""
+        from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+        import backend.app.models  # noqa: F401
+        from backend.app.core.database import Base
+        from backend.app.models.archive import PrintArchive
+        from backend.app.models.finance import CostCenter
+        from backend.app.models.print_queue import PrintQueueItem
+        from backend.app.models.printer import Printer
+        from backend.app.models.settings import Settings
+        from backend.app.models.user import User
+        from backend.app.services.print_scheduler import PrintScheduler
+
+        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+        session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+        async with session_maker() as db:
+            user = User(username="queue-user", email="queue-user@example.com", password_hash="x", role="user")
+            cost_center = CostCenter(name="Lab", is_active=True, is_private=False, monthly_budget=10.0)
+            printer = Printer(id=1, name="P1", serial_number="S1", ip_address="1.1.1.1", access_code="x")
+            billing_enabled = Settings(key="billing_enabled", value="true")
+            archive = PrintArchive(
+                id=1,
+                printer_id=1,
+                filename="test.3mf",
+                print_name="test",
+                file_path="missing.3mf",
+                file_size=1,
+                content_hash="hash",
+                status="completed",
+            )
+            db.add_all([user, cost_center, printer, archive, billing_enabled])
+            await db.flush()
+            item = PrintQueueItem(
+                printer_id=1,
+                archive_id=archive.id,
+                status="pending",
+                position=1,
+                created_by_id=user.id,
+                cost_center_id=cost_center.id,
+                estimated_cost=1.0,
+            )
+            db.add(item)
+            await db.commit()
+            item_id = item.id
+
+        scheduler = PrintScheduler()
+        async with session_maker() as db:
+            item = await db.get(PrintQueueItem, item_id)
+            with patch("backend.app.services.print_scheduler.printer_manager") as mock_pm:
+                await scheduler._start_print(db, item)
+                mock_pm.is_connected.assert_not_called()
+            await db.refresh(item)
+            assert item.status == "failed"
+            assert "cannot print with this cost center" in item.error_message
+
+        await engine.dispose()
+
+    @pytest.mark.asyncio
+    async def test_scheduler_budget_check_does_not_depend_on_billing_setting(self):
+        """Scheduler must still reject unauthorized cost centers when billing is disabled."""
+        from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+        import backend.app.models  # noqa: F401
+        from backend.app.core.database import Base
+        from backend.app.models.archive import PrintArchive
+        from backend.app.models.finance import CostCenter
+        from backend.app.models.print_queue import PrintQueueItem
+        from backend.app.models.printer import Printer
+        from backend.app.models.user import User
+        from backend.app.services.print_scheduler import PrintScheduler
+
+        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+        session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+        async with session_maker() as db:
+            user = User(username="queue-user", email="queue-user@example.com", password_hash="x", role="user")
+            cost_center = CostCenter(name="Lab", is_active=True, is_private=False, monthly_budget=10.0)
+            printer = Printer(id=1, name="P1", serial_number="S1", ip_address="1.1.1.1", access_code="x")
+            archive = PrintArchive(
+                id=1,
+                printer_id=1,
+                filename="test.3mf",
+                print_name="test",
+                file_path="missing.3mf",
+                file_size=1,
+                content_hash="hash",
+                status="completed",
+            )
+            db.add_all([user, cost_center, printer, archive])
+            await db.flush()
+            item = PrintQueueItem(
+                printer_id=1,
+                archive_id=archive.id,
+                status="pending",
+                position=1,
+                created_by_id=user.id,
+                cost_center_id=cost_center.id,
+                estimated_cost=1.0,
+            )
+            db.add(item)
+            await db.commit()
+            item_id = item.id
+
+        scheduler = PrintScheduler()
+        async with session_maker() as db:
+            item = await db.get(PrintQueueItem, item_id)
+            with patch("backend.app.services.print_scheduler.printer_manager") as mock_pm:
+                await scheduler._start_print(db, item)
+                mock_pm.is_connected.assert_not_called()
+            await db.refresh(item)
+            assert item.status == "failed"
+            assert "cannot print with this cost center" in item.error_message
+
+        await engine.dispose()

backend/tests/unit/test_printer_kill_switch.py

@@ -0,0 +1,197 @@
+from types import SimpleNamespace
+
+import pytest
+
+from backend.app import main as main_module
+
+
+@pytest.fixture(autouse=True)
+def clear_kill_switch_state():
+    main_module._unauthorized_print_kill_sent.clear()
+    main_module._expected_prints.clear()
+    main_module._active_prints.clear()
+    main_module._expected_print_registered_at.clear()
+    yield
+    main_module._unauthorized_print_kill_sent.clear()
+    main_module._expected_prints.clear()
+    main_module._active_prints.clear()
+    main_module._expected_print_registered_at.clear()
+
+
+def test_gcode_3mf_status_filename_matches_registered_expected_print():
+    state = SimpleNamespace(
+        current_print=None,
+        subtask_name="",
+        gcode_file="foreign_job.gcode.3mf",
+    )
+
+    keys = main_module._build_status_print_keys(7, state)
+
+    assert (7, "foreign_job.gcode.3mf") in keys
+    assert (7, "foreign_job.gcode") in keys
+
+
+@pytest.mark.asyncio
+async def test_unauthorized_active_print_triggers_stop(monkeypatch):
+    stop_calls: list[int] = []
+
+    async def fake_status(*args, **kwargs):
+        return None
+
+    async def kill_switch_enabled(_db):
+        return True
+
+    monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
+    monkeypatch.setattr(
+        main_module.printer_manager, "stop_print", lambda printer_id: stop_calls.append(printer_id) or True
+    )
+    monkeypatch.setattr(main_module.printer_manager, "get_printer", lambda printer_id: None)
+    monkeypatch.setattr(main_module.printer_manager, "get_model", lambda printer_id: None)
+    monkeypatch.setattr(main_module, "printer_state_to_dict", lambda state, printer_id, model: {})
+    monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status)
+    monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status)
+    monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled)
+
+    state = SimpleNamespace(
+        connected=True,
+        state="RUNNING",
+        progress=0,
+        layer_num=0,
+        temperatures={},
+        raw_data={},
+        stg_cur=0,
+        cooling_fan_speed=None,
+        big_fan1_speed=None,
+        big_fan2_speed=None,
+        chamber_light=False,
+        active_extruder=0,
+        tray_now=255,
+        door_open=False,
+        current_print=None,
+        subtask_name="foreign_job",
+        gcode_file="foreign_job.gcode",
+    )
+
+    await main_module.on_printer_status_change(7, state)
+
+    assert stop_calls == [7]
+    assert 7 in main_module._unauthorized_print_kill_sent
+
+
+@pytest.mark.asyncio
+async def test_bambuddy_authorized_print_is_not_stopped(monkeypatch):
+    monkeypatch.setitem(main_module._expected_prints, (7, "foreign_job"), 123)
+
+    stop_calls: list[int] = []
+
+    async def fake_status(*args, **kwargs):
+        return None
+
+    async def kill_switch_enabled(_db):
+        return True
+
+    monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
+    monkeypatch.setattr(
+        main_module.printer_manager, "stop_print", lambda printer_id: stop_calls.append(printer_id) or True
+    )
+    monkeypatch.setattr(main_module.printer_manager, "get_printer", lambda printer_id: None)
+    monkeypatch.setattr(main_module.printer_manager, "get_model", lambda printer_id: None)
+    monkeypatch.setattr(main_module, "printer_state_to_dict", lambda state, printer_id, model: {})
+    monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status)
+    monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status)
+    monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled)
+
+    state = SimpleNamespace(
+        connected=True,
+        state="RUNNING",
+        progress=0,
+        layer_num=0,
+        temperatures={},
+        raw_data={},
+        stg_cur=0,
+        cooling_fan_speed=None,
+        big_fan1_speed=None,
+        big_fan2_speed=None,
+        chamber_light=False,
+        active_extruder=0,
+        tray_now=255,
+        door_open=False,
+        current_print=None,
+        subtask_name="foreign_job",
+        gcode_file="foreign_job.gcode",
+    )
+
+    await main_module.on_printer_status_change(7, state)
+
+    assert stop_calls == []
+    assert 7 not in main_module._unauthorized_print_kill_sent
+
+
+@pytest.mark.asyncio
+async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch):
+    stop_calls: list[int] = []
+
+    async def fake_status(*args, **kwargs):
+        return None
+
+    async def kill_switch_enabled(_db):
+        return True
+
+    monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
+    monkeypatch.setattr(
+        main_module.printer_manager, "stop_print", lambda printer_id: stop_calls.append(printer_id) or True
+    )
+    monkeypatch.setattr(main_module.printer_manager, "get_printer", lambda printer_id: None)
+    monkeypatch.setattr(main_module.printer_manager, "get_model", lambda printer_id: None)
+    monkeypatch.setattr(main_module, "printer_state_to_dict", lambda state, printer_id, model: {})
+    monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status)
+    monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status)
+    monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled)
+
+    active_state = SimpleNamespace(
+        connected=True,
+        state="RUNNING",
+        progress=0,
+        layer_num=0,
+        temperatures={},
+        raw_data={},
+        stg_cur=0,
+        cooling_fan_speed=None,
+        big_fan1_speed=None,
+        big_fan2_speed=None,
+        chamber_light=False,
+        active_extruder=0,
+        tray_now=255,
+        door_open=False,
+        current_print=None,
+        subtask_name="foreign_job",
+        gcode_file="foreign_job.gcode",
+    )
+
+    idle_state = SimpleNamespace(
+        connected=True,
+        state="IDLE",
+        progress=0,
+        layer_num=0,
+        temperatures={},
+        raw_data={},
+        stg_cur=0,
+        cooling_fan_speed=None,
+        big_fan1_speed=None,
+        big_fan2_speed=None,
+        chamber_light=False,
+        active_extruder=0,
+        tray_now=255,
+        door_open=False,
+        current_print=None,
+        subtask_name="",
+        gcode_file=None,
+    )
+
+    await main_module.on_printer_status_change(7, active_state)
+    assert stop_calls == [7]
+    assert 7 in main_module._unauthorized_print_kill_sent
+
+    await main_module.on_printer_status_change(7, idle_state)
+
+    assert 7 not in main_module._unauthorized_print_kill_sent

frontend/src/__tests__/components/PrintModalBilling.test.tsx

@@ -0,0 +1,195 @@
+/**
+ * Tests for billing-related PrintModal request payloads.
+ */
+
+import React from 'react';
+import { BrowserRouter } from 'react-router-dom';
+import { QueryClient, QueryClientProvider } from '@tanstack/react-query';
+import { render, screen, waitFor } from '@testing-library/react';
+import userEvent from '@testing-library/user-event';
+import { http, HttpResponse } from 'msw';
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+
+import { server } from '../mocks/server';
+import { ThemeProvider } from '../../contexts/ThemeContext';
+
+const mockShowToast = vi.fn();
+const mockUseAuth = {
+  user: { id: 1, username: 'finance-user', permissions: ['cost_centers:read_own', 'printers:control'] },
+  authEnabled: true,
+  requiresSetup: false,
+  loading: false,
+  isAdmin: false,
+  login: vi.fn(),
+  loginWithToken: vi.fn(),
+  logout: vi.fn(),
+  refreshUser: vi.fn(),
+  refreshAuth: vi.fn(),
+  hasPermission: vi.fn((permission: string) => permission === 'cost_centers:read_own' || permission === 'printers:control'),
+  hasAnyPermission: vi.fn(() => true),
+  hasAllPermissions: vi.fn(() => true),
+  canModify: vi.fn(() => true),
+};
+
+vi.mock('../../contexts/AuthContext', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('../../contexts/AuthContext')>();
+  return {
+    ...actual,
+    useAuth: () => mockUseAuth,
+  };
+});
+
+vi.mock('../../contexts/ToastContext', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('../../contexts/ToastContext')>();
+  return {
+    ...actual,
+    useToast: () => ({ showToast: mockShowToast }),
+  };
+});
+
+import { PrintModal } from '../../components/PrintModal';
+
+function createTestQueryClient() {
+  return new QueryClient({
+    defaultOptions: {
+      queries: { retry: false, gcTime: 0 },
+      mutations: { retry: false },
+    },
+  });
+}
+
+function renderWithProviders(ui: React.ReactElement) {
+  const queryClient = createTestQueryClient();
+  return render(
+    <QueryClientProvider client={queryClient}>
+      <BrowserRouter>
+        <ThemeProvider>{ui}</ThemeProvider>
+      </BrowserRouter>
+    </QueryClientProvider>
+  );
+}
+
+const mockPrinters = [
+  { id: 1, name: 'X1 Carbon', model: 'X1C', ip_address: '192.168.1.100', enabled: true, is_active: true },
+];
+
+const mockQueueItem = {
+  id: 9,
+  printer_id: 1,
+  archive_id: 1,
+  position: 1,
+  scheduled_time: null,
+  require_previous_success: false,
+  auto_off_after: false,
+  gcode_injection: false,
+  manual_start: false,
+  ams_mapping: null,
+  plate_id: null,
+  bed_levelling: true,
+  flow_cali: false,
+  vibration_cali: true,
+  layer_inspect: false,
+  timelapse: false,
+  use_ams: true,
+  status: 'pending',
+  started_at: null,
+  completed_at: null,
+  error_message: null,
+  created_at: '2024-01-01T00:00:00Z',
+  archive_name: 'Billing Print',
+  archive_thumbnail: null,
+  printer_name: 'X1 Carbon',
+  print_time_seconds: 3600,
+  batch_id: null,
+  batch_name: null,
+  cost_center_id: 42,
+  estimated_cost: 12.34,
+};
+
+describe('PrintModal billing payloads', () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    server.use(
+      http.get('/api/v1/settings/', () => {
+        return HttpResponse.json({
+          currency: 'USD',
+          default_filament_cost: 25,
+          billing_enabled: true,
+          default_bed_levelling: true,
+          default_flow_cali: false,
+          default_vibration_cali: true,
+          default_layer_inspect: false,
+          default_timelapse: false,
+          stagger_group_size: 2,
+          stagger_interval_minutes: 5,
+          per_printer_mapping_expanded: false,
+          date_format: 'system',
+          time_format: 'system',
+        });
+      }),
+      http.get('/api/v1/printers/', () => HttpResponse.json(mockPrinters)),
+      http.get('/api/v1/printers/:id/status', () => {
+        return HttpResponse.json({ connected: true, state: 'IDLE', ams: [], vt_tray: [] });
+      }),
+      http.get('/api/v1/archives/:id', () => {
+        return HttpResponse.json({ id: 1, sliced_for_model: null });
+      }),
+      http.get('/api/v1/archives/:id/plates', () => {
+        return HttpResponse.json({ is_multi_plate: false, plates: [] });
+      }),
+      http.get('/api/v1/archives/:id/filament-requirements', () => {
+        return HttpResponse.json({ filaments: [] });
+      }),
+      http.get('/api/v1/finance/cost-centers/mine', () => {
+        return HttpResponse.json([
+          {
+            id: 42,
+            name: 'Lab',
+            is_private: false,
+            owner_user_id: null,
+            is_active: true,
+            total_balance: 0,
+            total_budget: 100,
+            monthly_budget: 100,
+            budget_mode: 'monthly',
+            budget_limit: 100,
+            budget_used: 0,
+            budget_available: 88,
+            can_print: true,
+          },
+        ]);
+      }),
+      http.patch('/api/v1/queue/:id', async ({ request }) => {
+        const body = await request.json() as Record<string, unknown>;
+        expect(body.cost_center_id).toBe(42);
+        expect(body.estimated_cost).toBe(12.34);
+        return HttpResponse.json({ id: 9, status: 'pending' });
+      })
+    );
+  });
+
+  it('includes the selected cost center and estimate when saving an edit-queue item', async () => {
+    const user = userEvent.setup();
+
+    renderWithProviders(
+      <PrintModal
+        mode="edit-queue-item"
+        archiveId={1}
+        archiveName="Billing Print"
+        queueItem={mockQueueItem as never}
+        onClose={vi.fn()}
+        onSuccess={vi.fn()}
+      />
+    );
+
+    await waitFor(() => {
+      expect(screen.getByText('Lab')).not.toBeNull();
+    });
+
+    await user.click(screen.getByRole('button', { name: /save/i }));
+
+    await waitFor(() => {
+      expect(mockShowToast).toHaveBeenCalled();
+    });
+  });
+});