diff --git a/backend/app/api/routes/finance.py b/backend/app/api/routes/finance.py index 8981f4edb..fb162dc61 100644 --- a/backend/app/api/routes/finance.py +++ b/backend/app/api/routes/finance.py @@ -3,7 +3,7 @@ from datetime import datetime, timezone from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from fastapi import APIRouter, Depends, HTTPException, Query -from sqlalchemy import and_, case, func, or_, select +from sqlalchemy import case, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload @@ -39,6 +39,11 @@ from backend.app.schemas.finance import ( WalletTransactionListResponse, WalletTransactionResponse, ) +from backend.app.services.finance_balance import ( + is_personal_transaction, + personal_balance_condition, + sync_personal_wallet_balance, +) router = APIRouter(prefix="/finance", tags=["finance"]) @@ -306,35 +311,21 @@ async def _build_personal_balance_map(db: AsyncSession, user_id: int) -> dict[in select( WalletTransaction.id, WalletTransaction.amount, - WalletTransaction.cost_center_id, - CostCenter.is_private, - CostCenter.owner_user_id, - ) - .where( - WalletTransaction.user_id == user_id, ) .outerjoin(CostCenter, WalletTransaction.cost_center_id == CostCenter.id) + .where(WalletTransaction.user_id == user_id, personal_balance_condition(user_id)) .order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc()) ) running_balance = 0.0 balance_map: dict[int, float] = {} - for transaction_id, amount, cost_center_id, is_private, owner_user_id in result.all(): - is_personal_cost_center = bool(cost_center_id is not None and is_private and owner_user_id == user_id) - if cost_center_id is None or is_personal_cost_center: - running_balance += float(amount) - balance_map[int(transaction_id)] = running_balance + for transaction_id, amount in result.all(): + running_balance += float(amount) + balance_map[int(transaction_id)] = running_balance return balance_map -def _personal_balance_condition(user_id: int): - return or_( - WalletTransaction.cost_center_id.is_(None), - and_(CostCenter.is_private.is_(True), CostCenter.owner_user_id == user_id), - ) - - async def _create_wallet_adjustment( db: AsyncSession, *, @@ -351,6 +342,7 @@ async def _create_wallet_adjustment( await _get_cost_center_or_404(db, cost_center_id) wallet = await _get_or_create_wallet(db, target_user_id) + affects_personal_wallet = await is_personal_transaction(db, target_user_id, cost_center_id) # Calculate balance_after for this specific transaction context if cost_center_id is None: @@ -358,7 +350,6 @@ async def _create_wallet_adjustment( new_balance = wallet.balance + amount if new_balance < 0: raise HTTPException(status_code=400, detail="Insufficient balance for withdrawal") - wallet.balance = new_balance balance_after = new_balance else: # Cost-center transaction: validate against cost center balance only (global, not per-user) @@ -372,7 +363,8 @@ async def _create_wallet_adjustment( if new_cc_balance < 0: raise HTTPException(status_code=400, detail="Insufficient cost center balance for withdrawal") balance_after = new_cc_balance - # Do NOT update wallet.balance for cost-center transactions + if affects_personal_wallet and wallet.balance + amount < 0: + raise HTTPException(status_code=400, detail="Insufficient balance for withdrawal") tx = WalletTransaction( user_id=target_user_id, @@ -385,12 +377,13 @@ async def _create_wallet_adjustment( ) db.add(tx) await db.flush() + await sync_personal_wallet_balance(db, wallet) await db.commit() await db.refresh(wallet) await db.refresh(tx) # Return appropriate balance based on transaction type - if cost_center_id is None: + if affects_personal_wallet: # Personal transaction: return user wallet balance response_balance = _to_balance_response(wallet) else: @@ -416,19 +409,7 @@ async def get_my_balance( """Return the current user's wallet balance.""" user = await _require_authenticated_user(current_user) wallet = await _get_or_create_wallet(db, user.id) - personal_balance_result = await db.execute( - select(func.coalesce(func.sum(WalletTransaction.amount), 0.0)) - .select_from(WalletTransaction) - .outerjoin(CostCenter, WalletTransaction.cost_center_id == CostCenter.id) - .where(WalletTransaction.user_id == user.id, _personal_balance_condition(user.id)) - ) - personal_balance = float(personal_balance_result.scalar_one() or 0.0) - return WalletBalanceResponse( - user_id=user.id, - balance=personal_balance, - currency=wallet.currency, - updated_at=wallet.updated_at, - ) + return _to_balance_response(wallet) @router.get("/me/transactions", response_model=WalletTransactionListResponse) @@ -504,47 +485,10 @@ async def get_all_transactions( async def _rebuild_wallet_ledger_for_user(db: AsyncSession, user_id: int) -> None: - """Recompute `balance_after` for all wallet transactions of a user. + """Rebuild through the same canonical ledger repair used at startup.""" + from backend.app.core.database import repair_wallet_ledger_internal - - Personal transactions (cost_center_id=None): running balance per user - - Cost-center transactions: running balance GLOBAL for entire cost center (not per-user) - - Also updates the user's wallet balance (sum of personal transactions only) - """ - result = await db.execute( - select(WalletTransaction) - .where(WalletTransaction.user_id == user_id) - .order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc()) - ) - user_transactions = result.scalars().all() - - # Handle personal transactions (cost_center_id=None) - personal_balance = 0.0 - for tx in user_transactions: - if tx.cost_center_id is None: - personal_balance += float(tx.amount) - tx.balance_after = personal_balance - db.add(tx) - - for cc_id in {tx.cost_center_id for tx in user_transactions if tx.cost_center_id is not None}: - # Get all transactions for this cost center (all users, all time) - result_all_cc = await db.execute( - select(WalletTransaction) - .where(WalletTransaction.cost_center_id == cc_id) - .order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc()) - ) - all_cc_transactions = result_all_cc.scalars().all() - - running = 0.0 - for tx in all_cc_transactions: - running += float(tx.amount) - tx.balance_after = running - db.add(tx) - - # Update user wallet balance (sum of all personal transactions only) - wallet = await _get_or_create_wallet(db, user_id) - wallet.balance = personal_balance - await db.flush() - await db.commit() + await repair_wallet_ledger_internal(db) @router.delete("/transactions/{transaction_id}") @@ -794,51 +738,19 @@ async def rebuild_balance_ledger( """Recompute balance_after for all wallet transactions. This rebuilds the running balance for all users and cost centers. - - Personal transactions: per-user running balance + - Personal transactions: unassigned plus the user's own private cost center - Cost-center transactions: global running balance for the entire cost center """ await _require_authenticated_user(current_user) - # Get ALL transactions sorted by timestamp - result = await db.execute( - select(WalletTransaction).order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc()) - ) - all_transactions = result.scalars().all() + from backend.app.core.database import repair_wallet_ledger_internal - # Build running balances per (user, cost_center_id) pair - # For each cost center, track its global running balance - # For each user's personal balance, track that separately - cc_running_balances: dict[int, float] = {} # cost_center_id -> running balance - user_personal_balances: dict[int, float] = {} # user_id -> personal running balance - - tx_updates: list[tuple[WalletTransaction, float]] = [] - - for tx in all_transactions: - if tx.cost_center_id is None: - # Personal transaction: per-user running balance - current = user_personal_balances.get(tx.user_id, 0.0) - new_balance = current + float(tx.amount) - user_personal_balances[tx.user_id] = new_balance - tx_updates.append((tx, new_balance)) - else: - # Cost-center transaction: global running balance for this cost center - current = cc_running_balances.get(tx.cost_center_id, 0.0) - new_balance = current + float(tx.amount) - cc_running_balances[tx.cost_center_id] = new_balance - tx_updates.append((tx, new_balance)) - - # Update all transactions with the new balance_after values - for tx, new_balance in tx_updates: - tx.balance_after = new_balance - db.add(tx) - - await db.flush() - await db.commit() + rebuilt = await repair_wallet_ledger_internal(db) return { "status": "success", - "transactions_rebuilt": len(all_transactions), - "message": f"Rebuilt balance_after for {len(all_transactions)} transactions", + "transactions_rebuilt": rebuilt, + "message": f"Rebuilt {rebuilt} wallet ledger values", } @@ -1053,6 +965,12 @@ async def delete_cost_center( raise HTTPException(status_code=400, detail="Cost center can only be deleted when balance is 0") await db.delete(center) + await db.flush() + # ON DELETE SET NULL reclassifies the former shared-center entries as + # personal transactions, so synchronize affected wallets immediately. + from backend.app.core.database import repair_wallet_ledger_internal + + await repair_wallet_ledger_internal(db) await db.commit() return {"status": "success"} diff --git a/backend/app/api/routes/print_queue.py b/backend/app/api/routes/print_queue.py index 3c1101767..bd23f7455 100644 --- a/backend/app/api/routes/print_queue.py +++ b/backend/app/api/routes/print_queue.py @@ -51,6 +51,7 @@ from backend.app.services.print_batch import ( load_progress, refresh_batch_status, ) +from backend.app.services.print_cost_estimate import estimate_queue_source_cost from backend.app.utils.printer_models import ( is_gcode_compatible, ) @@ -279,6 +280,55 @@ async def _resolve_source_path(db: AsyncSession, item: PrintQueueItem) -> Path | return None +async def _trusted_item_estimated_cost( + db: AsyncSession, + item: PrintQueueItem, + *, + printer_id: int | None, + plate_id: int | None, + ams_mapping: list[int] | str | None, +) -> float | None: + """Recompute an existing queue item's cost from its persisted source.""" + + if item.archive_id: + archive = await db.scalar(select(PrintArchive).where(PrintArchive.id == item.archive_id)) + return await estimate_queue_source_cost( + db, + archive=archive, + plate_id=plate_id, + ams_mapping=ams_mapping, + printer_id=printer_id, + ) + if item.library_file_id: + library_file = await db.scalar(LibraryFile.active().where(LibraryFile.id == item.library_file_id)) + return await estimate_queue_source_cost( + db, + library_file=library_file, + plate_id=plate_id, + ams_mapping=ams_mapping, + printer_id=printer_id, + ) + + result = await db.execute( + select(PrintQueueVariant, LibraryFile) + .join(LibraryFile, LibraryFile.id == PrintQueueVariant.library_file_id) + .where(PrintQueueVariant.queue_item_id == item.id) + ) + estimates = [ + await estimate_queue_source_cost( + db, + library_file=library_file, + plate_id=variant.plate_id, + ams_mapping=variant.ams_mapping, + printer_id=printer_id, + ) + for variant, library_file in result.all() + ] + if not estimates or any(cost is None for cost in estimates): + return None + return max(cost for cost in estimates if cost is not None) + + def _enrich_response(item: PrintQueueItem) -> PrintQueueItemResponse: """Add nested archive/printer/library_file info to response.""" # Parse ams_mapping from JSON string BEFORE model_validate @@ -945,10 +995,39 @@ async def add_to_queue( if not project_result.scalar_one_or_none(): raise HTTPException(status_code=404, detail="Project not found") + # Security boundary: the browser's estimated_cost is only a display hint. + # Budget enforcement and the persisted reservation value must be derived + # from the server-owned archive/library metadata and spool assignments. + if variant_specs: + variant_costs = [ + await estimate_queue_source_cost( + db, + library_file=variant_file, + plate_id=spec.plate_id, + ams_mapping=spec.ams_mapping, + printer_id=data.printer_id, + ) + for spec, variant_file, _model in variant_specs + ] + trusted_estimated_cost = ( + max(cost for cost in variant_costs if cost is not None) + if variant_costs and all(cost is not None for cost in variant_costs) + else None + ) + else: + trusted_estimated_cost = await estimate_queue_source_cost( + db, + archive=archive, + library_file=library_file, + plate_id=data.plate_id, + ams_mapping=data.ams_mapping, + printer_id=data.printer_id, + ) + await validate_print_budget( db, cost_center_id=data.cost_center_id, - estimated_cost=data.estimated_cost, + estimated_cost=trusted_estimated_cost, current_user=current_user, quantity=quantity, ) @@ -1010,7 +1089,7 @@ async def add_to_queue( archive_id=data.archive_id, library_file_id=data.library_file_id, cost_center_id=data.cost_center_id, - estimated_cost=data.estimated_cost, + estimated_cost=trusted_estimated_cost, scheduled_time=data.scheduled_time, require_previous_success=data.require_previous_success, auto_off_after=data.auto_off_after, @@ -1161,16 +1240,25 @@ async def bulk_update_queue_items( skipped_count += 1 continue + item_update_data = update_data.copy() if validates_billing_fields: + trusted_estimated_cost = await _trusted_item_estimated_cost( + db, + item, + printer_id=item_update_data.get("printer_id", item.printer_id), + plate_id=item.plate_id, + ams_mapping=item.ams_mapping, + ) + item_update_data["estimated_cost"] = trusted_estimated_cost await validate_print_budget( db, - cost_center_id=update_data.get("cost_center_id", item.cost_center_id), - estimated_cost=update_data.get("estimated_cost", item.estimated_cost), + cost_center_id=item_update_data.get("cost_center_id", item.cost_center_id), + estimated_cost=trusted_estimated_cost, current_user=user, exclude_queue_item_id=item.id, ) - for field, value in update_data.items(): + for field, value in item_update_data.items(): setattr(item, field, value) updated_count += 1 @@ -1826,10 +1914,19 @@ async def update_queue_item( json.dumps(update_data["nozzle_mapping"]) if update_data["nozzle_mapping"] else None ) + trusted_estimated_cost = await _trusted_item_estimated_cost( + db, + item, + printer_id=update_data.get("printer_id", item.printer_id), + plate_id=update_data.get("plate_id", item.plate_id), + ams_mapping=update_data.get("ams_mapping", item.ams_mapping), + ) + update_data["estimated_cost"] = trusted_estimated_cost + await validate_print_budget( db, cost_center_id=update_data.get("cost_center_id", item.cost_center_id), - estimated_cost=update_data.get("estimated_cost", item.estimated_cost), + estimated_cost=trusted_estimated_cost, current_user=user, exclude_queue_item_id=item.id, ) @@ -2169,6 +2266,14 @@ async def start_queue_item( if item.status != "pending": raise HTTPException(400, f"Can only start pending items, current status: '{item.status}'") + item.estimated_cost = await _trusted_item_estimated_cost( + db, + item, + printer_id=item.printer_id, + plate_id=item.plate_id, + ams_mapping=item.ams_mapping, + ) + await validate_print_budget( db, cost_center_id=item.cost_center_id, diff --git a/backend/app/core/database.py b/backend/app/core/database.py index d50855f59..7a07cb1ec 100644 --- a/backend/app/core/database.py +++ b/backend/app/core/database.py @@ -1232,6 +1232,14 @@ async def _migrate_create_finance_indexes(conn) -> None: await _safe_execute(conn, statement) +async def _migrate_add_print_archive_cost_center(conn) -> None: + """Add the nullable cost-center link missing from pre-billing archives.""" + await _safe_execute( + conn, + "ALTER TABLE print_archives ADD COLUMN cost_center_id INTEGER REFERENCES cost_centers(id) ON DELETE SET NULL", + ) + + async def run_migrations(conn): """Run all schema migrations and data backfills on startup. @@ -1585,6 +1593,9 @@ async def run_migrations(conn): except (OperationalError, ProgrammingError): pass # Already applied + # Migration: Add cost_center_id column to print_archives for billing metadata + await _migrate_add_print_archive_cost_center(conn) + # Migration: Add wiki_url column to maintenance_types for documentation links await _safe_execute(conn, "ALTER TABLE maintenance_types ADD COLUMN wiki_url VARCHAR(500)") @@ -4913,21 +4924,11 @@ async def repair_wallet_ledger(): This is called during database initialization to ensure all balance_after values are correct after code changes. Fixes old balance_after values to: - - Personal transactions: per-user running balance + - Personal transactions: unassigned plus the user's own private cost center - Cost-center transactions: global running balance for the entire cost center - Also updates UserWallet.balance to match final personal transaction balance. + Also updates UserWallet.balance to the canonical personal ledger sum. """ - from sqlalchemy import select - - from backend.app.models.finance import WalletTransaction - async with async_session() as session: - # Check if there are any transactions first - result = await session.execute(select(WalletTransaction).limit(1)) - if not result.scalar_one_or_none(): - logger.info("No wallet transactions found, skipping ledger rebuild") - return - updated_count = await repair_wallet_ledger_internal(session) await session.commit() @@ -4942,7 +4943,8 @@ async def repair_wallet_ledger_internal(session: AsyncSession): """ from sqlalchemy import select - from backend.app.models.finance import UserWallet, WalletTransaction + from backend.app.models.finance import CostCenter, UserWallet, WalletTransaction + from backend.app.services.finance_balance import transaction_affects_personal_balance # Get ALL transactions sorted by timestamp result = await session.execute( @@ -4950,8 +4952,10 @@ async def repair_wallet_ledger_internal(session: AsyncSession): ) all_transactions = result.scalars().all() - if not all_transactions: - return 0 + center_rows = await session.execute(select(CostCenter.id, CostCenter.is_private, CostCenter.owner_user_id)) + centers = { + int(center_id): (bool(is_private), owner_user_id) for center_id, is_private, owner_user_id in center_rows + } # Build running balances per (user, cost_center_id) pair cc_running_balances: dict[int, float] = {} # cost_center_id -> running balance @@ -4960,6 +4964,13 @@ async def repair_wallet_ledger_internal(session: AsyncSession): tx_updates: list[tuple[WalletTransaction, float]] = [] for tx in all_transactions: + center_is_private, center_owner_user_id = centers.get(tx.cost_center_id, (False, None)) + affects_personal = transaction_affects_personal_balance( + tx.user_id, + tx.cost_center_id, + is_private=center_is_private, + owner_user_id=center_owner_user_id, + ) if tx.cost_center_id is None: # Personal transaction: per-user running balance current = user_personal_balances.get(tx.user_id, 0.0) @@ -4973,6 +4984,11 @@ async def repair_wallet_ledger_internal(session: AsyncSession): cc_running_balances[tx.cost_center_id] = new_balance tx_updates.append((tx, new_balance)) + # The owner's private cost center is also part of that user's + # personal wallet. Shared centers never enter this sum. + if affects_personal: + user_personal_balances[tx.user_id] = user_personal_balances.get(tx.user_id, 0.0) + float(tx.amount) + # Update all transactions with the new balance_after values updated_count = 0 for tx, new_balance in tx_updates: @@ -4981,11 +4997,12 @@ async def repair_wallet_ledger_internal(session: AsyncSession): session.add(tx) updated_count += 1 - # Update UserWallet balances to match final personal balances - for user_id, balance in user_personal_balances.items(): - wallet_result = await session.execute(select(UserWallet).where(UserWallet.user_id == user_id)) - wallet = wallet_result.scalar_one_or_none() - if wallet and wallet.balance != balance: + # Update every wallet, including stale wallets whose canonical balance is + # now zero because their last personal transaction was deleted. + wallet_result = await session.execute(select(UserWallet)) + for wallet in wallet_result.scalars().all(): + balance = round(user_personal_balances.get(wallet.user_id, 0.0), 2) + if wallet.balance != balance: wallet.balance = balance session.add(wallet) updated_count += 1 diff --git a/backend/app/main.py b/backend/app/main.py index 21c4b2412..65129ffcd 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -106,6 +106,7 @@ from backend.app.services.mqtt_relay import mqtt_relay from backend.app.services.mqtt_smart_plug import mqtt_smart_plug_service from backend.app.services.notification_service import notification_service from backend.app.services.obico_detection import obico_detection_service +from backend.app.services.print_cost_estimate import plate_scoped_run_estimate as _plate_scoped_run_estimate from backend.app.services.print_scheduler import scheduler as print_scheduler from backend.app.services.printer_manager import ( init_printer_connections, @@ -425,6 +426,11 @@ _first_layer_notified: dict[int, bool] = {} # Track whether we already sent a kill-switch stop for the current unauthorized print _unauthorized_print_kill_sent: set[int] = set() +# Provider notification started when the kill switch stops a print. The later +# MQTT print-complete callback awaits this task and only sends its regular +# provider notification when the immediate attempt failed. +_kill_switch_notification_tasks: dict[int, asyncio.Task[bool]] = {} + # Track HMS errors that have been notified: {printer_id: set of error codes} # This prevents sending duplicate notifications for the same error _notified_hms_errors: dict[int, set[str]] = {} @@ -689,14 +695,94 @@ def _build_status_print_keys(printer_id: int, state: PrinterState) -> list[tuple return possible_keys -def _is_bambuddy_authorized_print(printer_id: int, state: PrinterState) -> bool: - """Return True when the current status belongs to a print started by Bambuddy.""" +async def _is_bambuddy_authorized_print(printer_id: int, state: PrinterState, db) -> bool | None: + """Resolve whether the current print was started by Bambuddy. + + ``None`` means identity is not yet safe to decide. The kill switch must + defer in that case: stopping a print is irreversible, and the first status + frames after a restart may arrive before all subtask fields are populated. + """ if printer_manager.get_current_print_user(printer_id): return True possible_keys = _build_status_print_keys(printer_id, state) - return any(key in _expected_prints or key in _active_prints for key in possible_keys) + if any(key in _expected_prints or key in _active_prints for key in possible_keys): + return True + + # In-memory ownership is lost on every Bambuddy restart. The archive row is + # the durable source of truth; subtask_id is minted per print and avoids + # authorizing an unrelated job that happens to reuse the same filename. + raw_subtask_id = getattr(state, "subtask_id", None) + subtask_id = str(raw_subtask_id).strip() if raw_subtask_id is not None else "" + if subtask_id in ("", "0"): + return None + + from backend.app.models.archive import PrintArchive + + result = await db.execute( + select(PrintArchive) + .where( + PrintArchive.printer_id == printer_id, + PrintArchive.status == "printing", + PrintArchive.subtask_id == subtask_id, + ) + .order_by(PrintArchive.created_at.desc()) + .limit(1) + ) + archive = result.scalar_one_or_none() + if archive is None: + return False + + # Rehydrate the fast in-memory path for subsequent status frames. Include + # both the archive filename and every normalized key reported by MQTT. + _active_prints[(printer_id, archive.filename)] = archive.id + for key in possible_keys: + _active_prints[key] = archive.id + return True + + +async def _send_kill_switch_provider_notification( + printer_id: int, + printer_name: str, + data: dict, +) -> bool: + """Send the immediate print-stopped provider notification. + + Returning a success flag lets the normal MQTT completion path retry when + this early notification could not be delivered. + """ + + logger = logging.getLogger(__name__) + try: + async with async_session() as db: + await notification_service.on_print_complete( + printer_id, + printer_name, + "stopped", + data, + db, + ) + return True + except Exception as e: + logger.warning( + "[KILL SWITCH] Immediate provider notification failed for printer %s: %s", + printer_id, + e, + ) + return False + + +async def _kill_switch_notification_already_sent(task: asyncio.Task[bool] | None) -> bool: + """Wait for an immediate kill-switch notification, if one was scheduled.""" + + if task is None: + return False + try: + return await task + except Exception as e: + logging.getLogger(__name__).warning("[KILL SWITCH] Notification task failed: %s", e) + return False async def _get_plug_energy(plug, db) -> dict | None: @@ -888,41 +974,6 @@ def _compute_run_filament_grams( return None -def _plate_scoped_run_estimate(archive, full_path) -> tuple[float | None, float | None]: - """Per-run (grams, cost) scoped to the plate this run actually printed (#2614). - - ``PrintArchive.filament_used_grams`` / ``.cost`` are the sum over EVERY plate of - the source 3MF — correct for the archive card and project rollup, but wrong for a - single plate dispatched from a multi-plate file: without scoping, each printed - plate of a 22-plate file logs the whole ~12 kg and inflates every statistic. When - the archive carries a ``plate_id`` and its 3MF is on disk, return that plate's - slicer estimate instead; cost is scaled by the plate's share of the whole so it - stays consistent with the scoped grams without re-doing the filament price lookup. - Falls back to the archive's whole-file values when there's no plate to scope to. - """ - whole_grams = archive.filament_used_grams - if archive.plate_id is None or full_path is None or not full_path.exists(): - return whole_grams, archive.cost - try: - from backend.app.utils.threemf_tools import extract_plate_metadata_from_3mf - - plate_grams = extract_plate_metadata_from_3mf(full_path, archive.plate_id).filament_used_grams - except Exception as exc: - logging.getLogger(__name__).debug( - "[#2614] plate-scoped estimate failed for archive %s (plate %s): %s", - archive.id, - archive.plate_id, - exc, - ) - return whole_grams, archive.cost - if not plate_grams or plate_grams <= 0: - return whole_grams, archive.cost - plate_cost = archive.cost - if archive.cost and whole_grams and whole_grams > 0: - plate_cost = round(archive.cost * (plate_grams / whole_grams), 2) - return round(plate_grams, 2), plate_cost - - def _get_start_ams_mapping(data: dict, archive_id: int | None) -> list[int] | None: """Resolve AMS mapping for print start without consuming stored queue/reprint state.""" stored_ams_mapping = data.get("ams_mapping") @@ -1380,17 +1431,31 @@ async def on_printer_status_change(printer_id: int, state: PrinterState): _unauthorized_print_kill_sent.discard(printer_id) else: kill_switch_enabled = False + authorization: bool | None = None status_logger = logging.getLogger(__name__) try: async with async_session() as db: from backend.app.services.finance_budget import is_printer_kill_switch_enabled kill_switch_enabled = await is_printer_kill_switch_enabled(db) + if kill_switch_enabled: + authorization = await _is_bambuddy_authorized_print(printer_id, state, db) except Exception as e: - status_logger.warning("[KILL SWITCH] Failed to read kill-switch setting for printer %s: %s", printer_id, e) + # Fail safe: a database/reconciliation error must never turn into an + # irreversible stop of a print whose ownership is still unknown. + authorization = None + status_logger.warning( + "[KILL SWITCH] Failed to reconcile print authorization for printer %s: %s", printer_id, e + ) - if not kill_switch_enabled or _is_bambuddy_authorized_print(printer_id, state): + if not kill_switch_enabled or authorization is True: _unauthorized_print_kill_sent.discard(printer_id) + elif authorization is None: + _unauthorized_print_kill_sent.discard(printer_id) + status_logger.debug( + "[KILL SWITCH] Deferring authorization for printer %s until archive state is reconciled", + printer_id, + ) elif printer_id in _unauthorized_print_kill_sent: pass else: @@ -1398,11 +1463,43 @@ async def on_printer_status_change(printer_id: int, state: PrinterState): stopped = printer_manager.stop_print(printer_id) if stopped: _unauthorized_print_kill_sent.add(printer_id) + printer_info = printer_manager.get_printer(printer_id) + printer_name = printer_info.name if printer_info else f"Printer {printer_id}" + filename = state.subtask_name or state.gcode_file or state.current_print or "Unknown" + notification_data = { + "status": "stopped", + "filename": state.gcode_file or state.current_print or "", + "subtask_name": state.subtask_name or "", + "progress": state.progress, + "reason": "unauthorized_print", + } status_logger.warning( "[KILL SWITCH] Stopped unauthorized print on printer %s (state=%s)", printer_id, state.state, ) + try: + await ws_manager.broadcast( + { + "type": "kill_switch_triggered", + "printer_id": printer_id, + "printer_name": printer_name, + "filename": filename, + "reason": "unauthorized_print", + } + ) + except Exception as e: + status_logger.warning( + "[KILL SWITCH] WebSocket notification failed for printer %s: %s", printer_id, e + ) + + previous_task = _kill_switch_notification_tasks.pop(printer_id, None) + if previous_task is not None and not previous_task.done(): + previous_task.cancel() + _kill_switch_notification_tasks[printer_id] = spawn_background_task( + _send_kill_switch_provider_notification(printer_id, printer_name, notification_data), + name=f"kill-switch-notification-{printer_id}", + ) else: status_logger.warning( "[KILL SWITCH] Could not stop unauthorized print on printer %s (state=%s)", @@ -2583,6 +2680,7 @@ async def on_print_start(printer_id: int, data: dict): # Clear any stale user-stopped flag from previous print cycles _user_stopped_printers.discard(printer_id) + _kill_switch_notification_tasks.pop(printer_id, None) # #1721: drop any leftover pre-captured finish frame from a prior print # so a never-consumed cache entry can't bleed into the new print's photo. @@ -4252,16 +4350,12 @@ async def _upgrade_finish_photo_from_timelapse(archive_id: int, archive_dir: Pat async def on_print_running_observed(printer_id: int, data: dict): - """Restart-recovery: capture a fresh timelapse baseline for a print that - started before Bambuddy came up. + """Restart-recovery for a print that started before Bambuddy came up. bambu_mqtt.py suppresses ``on_print_start`` on the first RUNNING push after Bambuddy startup (#1304 guard, prevents duplicate archive - creation). Without that path, ``_capture_timelapse_baseline_at_start`` - never runs and ``_scan_for_timelapse_with_retries`` falls into its - "take baseline now" fallback at completion time — but by then the - printer has already uploaded the in-flight MP4, so the baseline - includes it and no diff ever matches (#1485 follow-up). + creation). This hook restores the persisted archive into ``_active_prints`` + and captures the timelapse baseline that normally hangs off print start. Fires once per session, in lieu of on_print_start when restart-recovery kicks in. The printer doesn't upload the timelapse until after PRINT @@ -4270,21 +4364,15 @@ async def on_print_running_observed(printer_id: int, data: dict): """ logger = logging.getLogger(__name__) - # Avoid double-capture: on_print_start may have run earlier in this - # Bambuddy process if the print started AFTER startup and we crashed - # later in the same session. (Realistically this can't happen — the - # MQTT client object would have been recreated — but the cheap guard - # is correct regardless.) - if printer_id in _timelapse_baselines: - logger.debug( - "[TIMELAPSE] on_print_running_observed: baseline already present for printer %s, skipping", - printer_id, - ) - return - async with async_session() as db: from backend.app.models.printer import Printer + state = printer_manager.get_status(printer_id) + if state is not None: + authorization = await _is_bambuddy_authorized_print(printer_id, state, db) + if authorization is True: + logger.info("[RESTART] Restored active Bambuddy print for printer %s", printer_id) + result = await db.execute(select(Printer).where(Printer.id == printer_id)) printer = result.scalar_one_or_none() if not printer: @@ -4294,6 +4382,15 @@ async def on_print_running_observed(printer_id: int, data: dict): ) return + # Avoid double-capture: ownership reconciliation above must still run when + # a baseline already exists, but the camera work itself is one-shot. + if printer_id in _timelapse_baselines: + logger.debug( + "[TIMELAPSE] on_print_running_observed: baseline already present for printer %s, skipping", + printer_id, + ) + return + await _capture_timelapse_baseline_at_start(printer, printer_id, logger) @@ -4872,6 +4969,11 @@ async def on_print_complete(printer_id: int, data: dict): logger.info("[CALLBACK] on_print_complete started for printer %s", printer_id) + # A kill-switch stop sends its provider notification immediately. Keep the + # task so the later notification path can await it and avoid a duplicate; + # if that immediate attempt failed, the regular completion path retries. + kill_switch_notification_task = _kill_switch_notification_tasks.pop(printer_id, None) + # Drop the 3MF download cache for this printer (#972). The print is over, # nothing else legitimately needs the bytes; keeping them would only risk # handing a stale file to the next print if it reuses the same name. @@ -5452,9 +5554,12 @@ async def on_print_complete(printer_id: int, data: dict): logger.info( "[NOTIFY-BG] Sending notification without archive: printer=%s, status=%s", printer_id, ps ) - await notification_service.on_print_complete( - printer_id, p_name, ps, data, db, archive_data=no_archive_data - ) + if not await _kill_switch_notification_already_sent(kill_switch_notification_task): + await notification_service.on_print_complete( + printer_id, p_name, ps, data, db, archive_data=no_archive_data + ) + else: + logger.info("[NOTIFY-BG] Skipped duplicate kill-switch provider notification") # Send user-specific email if we have a created_by_id if no_archive_data and no_archive_data.get("created_by_id"): @@ -6134,9 +6239,12 @@ async def on_print_complete(printer_id: int, data: dict): except Exception as e: logger.warning("[NOTIFY-BG] Failed to read finish photo bytes: %s", e) - await notification_service.on_print_complete( - printer_id, printer_name, print_status, data, db, archive_data=archive_data - ) + if not await _kill_switch_notification_already_sent(kill_switch_notification_task): + await notification_service.on_print_complete( + printer_id, printer_name, print_status, data, db, archive_data=archive_data + ) + else: + logger.info("[NOTIFY-BG] Skipped duplicate kill-switch provider notification") # Send user-specific email notification if archive_data: diff --git a/backend/app/services/finance_balance.py b/backend/app/services/finance_balance.py new file mode 100644 index 000000000..b3fee3f13 --- /dev/null +++ b/backend/app/services/finance_balance.py @@ -0,0 +1,65 @@ +"""Canonical definition and synchronization of a user's personal balance.""" + +from sqlalchemy import and_, func, or_, select +from sqlalchemy.ext.asyncio import AsyncSession + +from backend.app.models.finance import CostCenter, UserWallet, WalletTransaction + + +def transaction_affects_personal_balance( + user_id: int, + cost_center_id: int | None, + *, + is_private: bool = False, + owner_user_id: int | None = None, +) -> bool: + """Apply the canonical definition to already-loaded transaction data.""" + + return cost_center_id is None or (is_private and owner_user_id == user_id) + + +def personal_balance_condition(user_id: int): + """Return the SQL condition for transactions in a personal wallet. + + Unassigned transactions and transactions assigned to the user's own + private cost center are personal. Shared cost centers are not. + """ + + return or_( + WalletTransaction.cost_center_id.is_(None), + and_(CostCenter.is_private.is_(True), CostCenter.owner_user_id == user_id), + ) + + +async def calculate_personal_balance(db: AsyncSession, user_id: int) -> float: + result = await db.execute( + select(func.coalesce(func.sum(WalletTransaction.amount), 0.0)) + .select_from(WalletTransaction) + .outerjoin(CostCenter, WalletTransaction.cost_center_id == CostCenter.id) + .where(WalletTransaction.user_id == user_id, personal_balance_condition(user_id)) + ) + return round(float(result.scalar_one() or 0.0), 2) + + +async def is_personal_transaction(db: AsyncSession, user_id: int, cost_center_id: int | None) -> bool: + if cost_center_id is None: + return True + result = await db.execute( + select(CostCenter.is_private, CostCenter.owner_user_id).where(CostCenter.id == cost_center_id) + ) + center = result.one_or_none() + if center is None: + return False + return transaction_affects_personal_balance( + user_id, + cost_center_id, + is_private=bool(center.is_private), + owner_user_id=center.owner_user_id, + ) + + +async def sync_personal_wallet_balance(db: AsyncSession, wallet: UserWallet) -> float: + balance = await calculate_personal_balance(db, wallet.user_id) + wallet.balance = balance + db.add(wallet) + return balance diff --git a/backend/app/services/finance_billing.py b/backend/app/services/finance_billing.py index de35b1e7e..365198ef0 100644 --- a/backend/app/services/finance_billing.py +++ b/backend/app/services/finance_billing.py @@ -6,6 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from backend.app.models.archive import PrintArchive from backend.app.models.finance import TransactionType, UserWallet, WalletTransaction +from backend.app.services.finance_balance import sync_personal_wallet_balance from backend.app.services.finance_budget import is_billing_enabled, release_budget_reservation logger = logging.getLogger(__name__) @@ -185,10 +186,6 @@ async def apply_print_charge_for_archive( await db.flush() logger.info(f"Created new wallet for user ID {archive.created_by_id}.") - # Persist wallet balances rounded to cents - new_wallet_balance = round(float(wallet.balance) - charge, 2) - wallet.balance = new_wallet_balance - label = archive.print_name or archive.filename or f"Archive {archive.id}" description = f"Print charge: {label}{' ' + reason_suffix if reason_suffix else ''}" @@ -219,6 +216,10 @@ async def apply_print_charge_for_archive( await db.rollback() return False + # Rebuild from the canonical personal-ledger definition. A shared cost + # center charge must not debit the user's personal wallet. + new_wallet_balance = await sync_personal_wallet_balance(db, wallet) + # Consume matching budget reservations after the transaction is persisted await release_budget_reservation(db, print_archive_id=archive.id, status="consumed") logger.info(f"Applied print charge for archive ID {archive_id}. New balance: {new_wallet_balance}.") diff --git a/backend/app/services/print_cost_estimate.py b/backend/app/services/print_cost_estimate.py new file mode 100644 index 000000000..88baf4c94 --- /dev/null +++ b/backend/app/services/print_cost_estimate.py @@ -0,0 +1,155 @@ +"""Trusted server-side cost estimates for queued prints.""" + +import json +import logging +from pathlib import Path + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + +from backend.app.core.config import settings +from backend.app.models.archive import PrintArchive +from backend.app.models.library import LibraryFile +from backend.app.models.spool_assignment import SpoolAssignment +from backend.app.utils import threemf_tools +from backend.app.utils.safe_path import safe_join_under + +logger = logging.getLogger(__name__) + + +def plate_scoped_run_estimate( + archive: PrintArchive, + full_path: Path | None, + plate_id: int | None = None, +) -> tuple[float | None, float | None]: + """Return trusted ``(grams, cost)`` for one run of an archive plate.""" + + whole_grams = archive.filament_used_grams + selected_plate = archive.plate_id if plate_id is None else plate_id + if selected_plate is None or full_path is None or not full_path.exists(): + return whole_grams, archive.cost + try: + plate_grams = threemf_tools.extract_plate_metadata_from_3mf(full_path, selected_plate).filament_used_grams + except Exception as exc: + logger.debug( + "Plate-scoped estimate failed for archive %s (plate %s): %s", + archive.id, + selected_plate, + exc, + ) + return whole_grams, archive.cost + if not plate_grams or plate_grams <= 0: + return whole_grams, archive.cost + plate_cost = archive.cost + if archive.cost and whole_grams and whole_grams > 0: + plate_cost = round(archive.cost * (plate_grams / whole_grams), 2) + return round(plate_grams, 2), plate_cost + + +def _source_path(library_file: LibraryFile) -> Path: + path = Path(library_file.file_path) + if path.is_absolute(): + # SEC-PATH-OK: absolute paths are persisted LibraryFile locations for + # configured external libraries; this branch performs no path join. + return path + return safe_join_under(settings.base_dir, library_file.file_path, http=False) + + +def _parse_mapping(mapping: list[int] | str | None) -> list[int] | None: + if isinstance(mapping, list): + return mapping + if isinstance(mapping, str): + try: + parsed = json.loads(mapping) + except (TypeError, json.JSONDecodeError): + return None + return parsed if isinstance(parsed, list) else None + return None + + +def _global_tray_id(assignment: SpoolAssignment) -> int: + if assignment.ams_id == 255: + return 254 + assignment.tray_id + if assignment.ams_id >= 128: + return assignment.ams_id + return assignment.ams_id * 4 + assignment.tray_id + + +async def _default_cost_per_kg(db: AsyncSession) -> float: + from backend.app.api.routes.settings import get_setting + + raw = await get_setting(db, "default_filament_cost") + try: + return float(raw) if raw is not None else 25.0 + except (TypeError, ValueError): + return 25.0 + + +async def estimate_queue_source_cost( + db: AsyncSession, + *, + archive: PrintArchive | None = None, + library_file: LibraryFile | None = None, + plate_id: int | None = None, + ams_mapping: list[int] | str | None = None, + printer_id: int | None = None, +) -> float | None: + """Compute a queue cost without trusting the request's display hint.""" + + if archive is not None: + archive_path = settings.base_dir / archive.file_path + _grams, cost = plate_scoped_run_estimate(archive, archive_path, plate_id) + return float(cost) if cost is not None and cost > 0 else None + + if library_file is None: + return None + + path = _source_path(library_file) + usage: list[dict] = [] + if path.exists(): + usage = threemf_tools.extract_plate_metadata_from_3mf(path, plate_id).filament_usage + + metadata = library_file.file_metadata or {} + if not usage: + try: + grams = float(metadata.get("filament_used_grams") or 0) + except (TypeError, ValueError): + grams = 0 + if grams > 0: + usage = [{"slot_id": 1, "used_g": grams}] + + if not usage: + return None + + default_cost = await _default_cost_per_kg(db) + cost_by_tray: dict[int, float | None] = {} + mapping = _parse_mapping(ams_mapping) + if printer_id is not None and mapping: + assignments = ( + ( + await db.execute( + select(SpoolAssignment) + .options(selectinload(SpoolAssignment.spool)) + .where(SpoolAssignment.printer_id == printer_id) + ) + ) + .scalars() + .all() + ) + cost_by_tray = {_global_tray_id(a): a.spool.cost_per_kg for a in assignments} + + total = 0.0 + for filament in usage: + try: + slot_id = int(filament.get("slot_id") or 0) + grams = float(filament.get("used_g") or 0) + except (TypeError, ValueError): + continue + tray_id = mapping[slot_id - 1] if mapping and 0 < slot_id <= len(mapping) else None + cost_per_kg = cost_by_tray.get(tray_id) if tray_id is not None else None + if cost_per_kg is None or cost_per_kg <= 0: + cost_per_kg = default_cost + total += (grams / 1000.0) * cost_per_kg + + return round(total, 2) if total > 0 else None diff --git a/backend/app/services/print_scheduler.py b/backend/app/services/print_scheduler.py index 0bf6d3f65..aecfcd829 100644 --- a/backend/app/services/print_scheduler.py +++ b/backend/app/services/print_scheduler.py @@ -43,6 +43,7 @@ from backend.app.services.finance_budget import ( ) from backend.app.services.ha_sensor_manager import ha_sensor_manager from backend.app.services.notification_service import notification_service +from backend.app.services.print_cost_estimate import estimate_queue_source_cost from backend.app.services.printer_manager import ( printer_manager, supports_airduct, @@ -3684,6 +3685,20 @@ class PrintScheduler: from backend.app.models.user import User queue_user = await db.get(User, item.created_by_id) if item.created_by_id is not None else None + # Recompute at the final authorization boundary as well as enqueue + # time. This covers rows created before the server-side estimate + # migration and prevents any alternate write path from weakening + # the budget reservation. + archive = await db.get(PrintArchive, item.archive_id) if item.archive_id is not None else None + library_file = await db.get(LibraryFile, item.library_file_id) if item.library_file_id is not None else None + item.estimated_cost = await estimate_queue_source_cost( + db, + archive=archive, + library_file=library_file, + plate_id=item.plate_id, + ams_mapping=item.ams_mapping, + printer_id=item.printer_id, + ) await validate_print_budget( db, cost_center_id=item.cost_center_id, diff --git a/backend/tests/integration/test_finance_api.py b/backend/tests/integration/test_finance_api.py index eff18dd04..832265b32 100644 --- a/backend/tests/integration/test_finance_api.py +++ b/backend/tests/integration/test_finance_api.py @@ -7,8 +7,10 @@ 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.group import Group 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 class TestFinanceAPI: @@ -143,34 +145,30 @@ class TestFinanceAPI: 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. - """ + """A user's private cost center and unassigned entries share one 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( + private_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 + assert private_center is not None - # Deposit to cost center: affects cost-center balance, not user wallet + # The owner's private cost center affects both its own ledger and the 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}, + json={"amount": 25.0, "description": "Initial CC top-up", "cost_center_id": private_center.id}, headers=auth_headers, ) assert deposit.status_code == 200 - assert deposit.json()["transaction"]["cost_center_id"] == shared_center.id + assert deposit.json()["transaction"]["cost_center_id"] == private_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 + # A withdrawal updates both views by the same amount. 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}, + json={"amount": 5.0, "description": "CC Usage", "cost_center_id": private_center.id}, headers=auth_headers, ) assert withdraw.status_code == 200 @@ -186,8 +184,8 @@ class TestFinanceAPI: ) 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 + assert personal_deposit.json()["transaction"]["balance_after"] == 50.0 + assert personal_deposit.json()["balance"]["balance"] == 50.0 transactions_response = await async_client.get( f"/api/v1/finance/users/{created_user['id']}/transactions", headers=auth_headers @@ -195,7 +193,7 @@ class TestFinanceAPI: 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] + cc_txs = [tx for tx in transactions if tx["cost_center_id"] == private_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 @@ -204,7 +202,7 @@ class TestFinanceAPI: 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 + assert balance_response.json()["balance"] == 50.0 # Delete personal transaction, user wallet should decrease personal_tx = next(tx for tx in transactions if tx["cost_center_id"] is None) @@ -217,7 +215,7 @@ class TestFinanceAPI: 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 + assert balance_after_delete.json()["balance"] == 20.0 remaining = await async_client.get( f"/api/v1/finance/users/{created_user['id']}/transactions", headers=auth_headers @@ -225,6 +223,74 @@ class TestFinanceAPI: assert remaining.status_code == 200 assert len(remaining.json()) == 2 # 2 CC transactions remain + @pytest.mark.asyncio + @pytest.mark.integration + async def test_admin_and_user_balances_agree_after_charge_adjustment_and_delete( + self, + async_client: AsyncClient, + auth_headers: dict[str, str], + db_session, + ): + created_user = await self._create_user_via_api(async_client, auth_headers, "balance-lifecycle") + user = await db_session.get(User, created_user["id"]) + operators = await db_session.scalar(select(Group).where(Group.name == "Operators")) + assert operators is not None + user.groups.append(operators) + await db_session.commit() + user_headers = await self._login_user(async_client, "balance-lifecycle") + 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 + + deposit = await async_client.post( + f"/api/v1/finance/users/{user.id}/deposit", + json={"amount": 20.0, "cost_center_id": private_center.id}, + headers=auth_headers, + ) + assert deposit.status_code == 200 + + archive = PrintArchive( + filename="charged.gcode.3mf", + file_path="archives/test/charged.gcode.3mf", + file_size=10, + status="completed", + cost=4.0, + created_by_id=user.id, + cost_center_id=private_center.id, + ) + db_session.add(archive) + await db_session.commit() + assert await apply_print_charge_for_archive(db_session, archive.id) is True + await db_session.commit() + + adjustment = await async_client.post( + f"/api/v1/finance/users/{user.id}/deposit", + json={"amount": 5.0, "description": "temporary adjustment"}, + headers=auth_headers, + ) + assert adjustment.status_code == 200 + + async def assert_views_agree(expected: float): + admin_view = await async_client.get( + f"/api/v1/finance/users/{user.id}/balance", + headers=auth_headers, + ) + user_view = await async_client.get("/api/v1/finance/me/balance", headers=user_headers) + assert admin_view.status_code == 200 + assert user_view.status_code == 200 + assert admin_view.json()["balance"] == expected + assert user_view.json()["balance"] == expected + + await assert_views_agree(21.0) + + delete_response = await async_client.delete( + f"/api/v1/finance/transactions/{adjustment.json()['transaction']['id']}", + headers=auth_headers, + ) + assert delete_response.status_code == 200 + await assert_views_agree(16.0) + @pytest.mark.asyncio @pytest.mark.integration async def test_delete_cost_center_transaction_rebuilds_remaining_ledger( @@ -283,7 +349,7 @@ class TestFinanceAPI: 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 + assert balance_response.json()["balance"] == 7.0 @pytest.mark.asyncio @pytest.mark.integration diff --git a/backend/tests/integration/test_print_queue_api.py b/backend/tests/integration/test_print_queue_api.py index 9dba98374..95b9fb792 100644 --- a/backend/tests/integration/test_print_queue_api.py +++ b/backend/tests/integration/test_print_queue_api.py @@ -146,7 +146,7 @@ class TestPrintQueueAPI: """Verify item can be added to queue with cost_center_id.""" await enable_billing(db_session) printer = await printer_factory() - archive = await archive_factory() + archive = await archive_factory(cost=1.25, filament_used_grams=50.0) cost_center = CostCenter(name="Queue CC", is_active=True, is_private=False) db_session.add(cost_center) await db_session.commit() @@ -158,7 +158,7 @@ class TestPrintQueueAPI: "printer_id": printer.id, "archive_id": archive.id, "cost_center_id": cost_center.id, - "estimated_cost": 1.25, + "estimated_cost": 0.01, }, ) assert response.status_code == 200 @@ -175,13 +175,13 @@ class TestPrintQueueAPI: @pytest.mark.asyncio @pytest.mark.integration - async def test_add_to_queue_with_cost_center_requires_estimated_cost( + async def test_add_to_queue_derives_cost_without_client_estimate( self, async_client: AsyncClient, printer_factory, archive_factory, db_session ): - """Cost-center queue items require an estimated cost for budget checks.""" + """The persisted budget estimate comes from the archive, not the request.""" await enable_billing(db_session) printer = await printer_factory() - archive = await archive_factory() + archive = await archive_factory(cost=1.25, filament_used_grams=50.0) 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() @@ -196,18 +196,18 @@ class TestPrintQueueAPI: }, ) - assert response.status_code == 400 - assert "Estimated cost is required" in response.json()["detail"] + assert response.status_code == 200 + assert response.json()["estimated_cost"] == 1.25 @pytest.mark.asyncio @pytest.mark.integration - async def test_add_to_queue_rejects_when_estimated_cost_exceeds_budget( + async def test_add_to_queue_rejects_tampered_client_cost_when_server_cost_exceeds_budget( self, async_client: AsyncClient, printer_factory, archive_factory, db_session ): - """Queue creation is rejected if the estimated cost exceeds remaining budget.""" + """A forged low client hint cannot bypass the server-derived budget check.""" await enable_billing(db_session) printer = await printer_factory() - archive = await archive_factory() + archive = await archive_factory(cost=2.0, filament_used_grams=50.0) 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() @@ -219,7 +219,7 @@ class TestPrintQueueAPI: "printer_id": printer.id, "archive_id": archive.id, "cost_center_id": cost_center.id, - "estimated_cost": 2.0, + "estimated_cost": 0.01, }, ) @@ -234,7 +234,7 @@ class TestPrintQueueAPI: """Billing enforcement rejects queue jobs that omit cost center.""" await enable_billing(db_session) printer = await printer_factory() - archive = await archive_factory() + archive = await archive_factory(cost=1.0, filament_used_grams=50.0) response = await async_client.post( "/api/v1/queue/", @@ -255,7 +255,7 @@ class TestPrintQueueAPI: """Open queue items reserve budget until they leave pending/printing states.""" await enable_billing(db_session) printer = await printer_factory() - archive = await archive_factory() + archive = await archive_factory(cost=3.0, filament_used_grams=50.0) 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() @@ -274,7 +274,7 @@ class TestPrintQueueAPI: "printer_id": printer.id, "archive_id": archive.id, "cost_center_id": cost_center.id, - "estimated_cost": 3.0, + "estimated_cost": 0.01, }, ) @@ -345,7 +345,7 @@ class TestPrintQueueAPI: """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() + archive = await archive_factory(cost=1.25, filament_used_grams=50.0) 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) @@ -354,7 +354,7 @@ class TestPrintQueueAPI: response = await async_client.patch( f"/api/v1/queue/{item.id}", - json={"cost_center_id": cost_center.id, "estimated_cost": 1.25}, + json={"cost_center_id": cost_center.id, "estimated_cost": 0.01}, ) assert response.status_code == 200 @@ -822,7 +822,7 @@ class TestPrintQueueAPI: """Deleting a pending cost-center queue item releases its reserved budget.""" await enable_billing(db_session) printer = await printer_factory() - archive = await archive_factory() + archive = await archive_factory(cost=3.0, filament_used_grams=50.0) cost_center = CostCenter( name="Delete Releases Budget CC", is_active=True, is_private=False, monthly_budget=10.0 ) @@ -843,7 +843,7 @@ class TestPrintQueueAPI: "printer_id": printer.id, "archive_id": archive.id, "cost_center_id": cost_center.id, - "estimated_cost": 3.0, + "estimated_cost": 0.01, }, ) assert blocked.status_code == 400 @@ -857,7 +857,7 @@ class TestPrintQueueAPI: "printer_id": printer.id, "archive_id": archive.id, "cost_center_id": cost_center.id, - "estimated_cost": 3.0, + "estimated_cost": 0.01, }, ) assert allowed.status_code == 200 @@ -1730,11 +1730,12 @@ class TestBulkUpdateEndpoint: @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 + self, async_client: AsyncClient, queue_item_factory, archive_factory, db_session ): - """Bulk updates cannot reserve more than the available cost-center budget.""" + """A forged bulk-update hint cannot weaken the server-derived reservation.""" await enable_billing(db_session) - item = await queue_item_factory() + archive = await archive_factory(cost=3.0, filament_used_grams=50.0) + item = await queue_item_factory(archive_id=archive.id) 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) @@ -1747,7 +1748,7 @@ class TestBulkUpdateEndpoint: response = await async_client.patch( "/api/v1/queue/bulk", - json={"item_ids": [item.id], "cost_center_id": cost_center.id, "estimated_cost": 3.0}, + json={"item_ids": [item.id], "cost_center_id": cost_center.id, "estimated_cost": 0.01}, ) assert response.status_code == 400 diff --git a/backend/tests/unit/services/test_finance_service_billing.py b/backend/tests/unit/services/test_finance_service_billing.py index 997bef8a5..a2a544bac 100644 --- a/backend/tests/unit/services/test_finance_service_billing.py +++ b/backend/tests/unit/services/test_finance_service_billing.py @@ -60,7 +60,7 @@ class TestFinanceBilling: wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id)) assert wallet is not None - assert wallet.balance == -7.5 + assert wallet.balance == 0.0 tx = await db_session.scalar(select(WalletTransaction).where(WalletTransaction.print_run_id == "run-1")) assert tx is not None @@ -86,7 +86,7 @@ class TestFinanceBilling: 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 + assert wallet.balance == 0.0 rows = ( (await db_session.execute(select(WalletTransaction).where(WalletTransaction.user_id == user.id))) @@ -277,7 +277,7 @@ class TestPartialPrintCharges: 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 == -3.0 + assert wallet.balance == 0.0 transaction = await db_session.scalar( select(WalletTransaction).where(WalletTransaction.print_archive_id == archive.id) @@ -374,7 +374,7 @@ class TestPartialPrintCharges: 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 + assert wallet.balance == 0.0 tx = await db_session.scalar( select(WalletTransaction) @@ -476,7 +476,7 @@ class TestPartialPrintCharges: 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 + assert wallet.balance == 0.0 @pytest.mark.asyncio async def test_completed_print_still_charges_full_cost(self, db_session): @@ -512,7 +512,7 @@ class TestPartialPrintCharges: 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 + assert wallet.balance == 0.0 @pytest.mark.asyncio async def test_partial_charge_with_cost_center_override(self, db_session): diff --git a/backend/tests/unit/services/test_print_cost_estimate.py b/backend/tests/unit/services/test_print_cost_estimate.py new file mode 100644 index 000000000..ff806b4ee --- /dev/null +++ b/backend/tests/unit/services/test_print_cost_estimate.py @@ -0,0 +1,85 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from backend.app.services import print_cost_estimate + + +@pytest.fixture +def library_file(tmp_path): + path = tmp_path / "queued.gcode.3mf" + path.write_bytes(b"stub") + return SimpleNamespace(file_path=str(path), file_metadata={}) + + +@pytest.mark.asyncio +async def test_library_estimate_uses_server_default_cost(monkeypatch, library_file): + monkeypatch.setattr( + print_cost_estimate.threemf_tools, + "extract_plate_metadata_from_3mf", + lambda *_args: SimpleNamespace( + filament_usage=[ + {"slot_id": 1, "used_g": 100.0}, + {"slot_id": 2, "used_g": 50.0}, + ] + ), + ) + monkeypatch.setattr(print_cost_estimate, "_default_cost_per_kg", AsyncMock(return_value=20.0)) + + cost = await print_cost_estimate.estimate_queue_source_cost( + SimpleNamespace(), + library_file=library_file, + plate_id=1, + ) + + assert cost == 3.0 + + +@pytest.mark.asyncio +async def test_library_estimate_uses_server_spool_assignment_costs(monkeypatch, library_file): + monkeypatch.setattr( + print_cost_estimate.threemf_tools, + "extract_plate_metadata_from_3mf", + lambda *_args: SimpleNamespace( + filament_usage=[ + {"slot_id": 1, "used_g": 100.0}, + {"slot_id": 2, "used_g": 50.0}, + ] + ), + ) + monkeypatch.setattr(print_cost_estimate, "_default_cost_per_kg", AsyncMock(return_value=20.0)) + + assignments = [ + SimpleNamespace(ams_id=0, tray_id=0, spool=SimpleNamespace(cost_per_kg=10.0)), + SimpleNamespace(ams_id=0, tray_id=1, spool=SimpleNamespace(cost_per_kg=30.0)), + ] + result = SimpleNamespace(scalars=lambda: SimpleNamespace(all=lambda: assignments)) + db = SimpleNamespace(execute=AsyncMock(return_value=result)) + + cost = await print_cost_estimate.estimate_queue_source_cost( + db, + library_file=library_file, + plate_id=1, + ams_mapping=[0, 1], + printer_id=7, + ) + + assert cost == 2.5 + + +@pytest.mark.asyncio +async def test_missing_server_metadata_does_not_fall_back_to_client_hint(monkeypatch, library_file): + monkeypatch.setattr( + print_cost_estimate.threemf_tools, + "extract_plate_metadata_from_3mf", + lambda *_args: SimpleNamespace(filament_usage=[]), + ) + + cost = await print_cost_estimate.estimate_queue_source_cost( + SimpleNamespace(), + library_file=library_file, + plate_id=1, + ) + + assert cost is None diff --git a/backend/tests/unit/test_finance_table_migration.py b/backend/tests/unit/test_finance_table_migration.py index 98088761d..d8ec7dd53 100644 --- a/backend/tests/unit/test_finance_table_migration.py +++ b/backend/tests/unit/test_finance_table_migration.py @@ -7,7 +7,11 @@ import pytest from sqlalchemy import text from sqlalchemy.ext.asyncio import create_async_engine -from backend.app.core.database import _migrate_create_finance_indexes, _migrate_create_finance_tables +from backend.app.core.database import ( + _migrate_add_print_archive_cost_center, + _migrate_create_finance_indexes, + _migrate_create_finance_tables, +) EXPECTED_TABLES = { "cost_centers", @@ -66,6 +70,31 @@ async def test_legacy_cost_center_indexes_are_delayed_until_columns_exist(): await engine.dispose() +@pytest.mark.asyncio +async def test_print_archive_cost_center_is_added_idempotently_on_sqlite(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + + try: + async with engine.begin() as conn: + await conn.execute(text("PRAGMA foreign_keys = ON")) + await conn.execute(text("CREATE TABLE cost_centers (id INTEGER PRIMARY KEY)")) + await conn.execute(text("CREATE TABLE print_archives (id INTEGER PRIMARY KEY)")) + + await _migrate_add_print_archive_cost_center(conn) + await _migrate_add_print_archive_cost_center(conn) + + columns = await conn.execute(text("PRAGMA table_info(print_archives)")) + foreign_keys = await conn.execute(text("PRAGMA foreign_key_list(print_archives)")) + + assert "cost_center_id" in {row[1] for row in columns} + assert any( + row[2] == "cost_centers" and row[3] == "cost_center_id" and row[6].upper() == "SET NULL" + for row in foreign_keys + ) + finally: + await engine.dispose() + + @pytest.mark.asyncio async def test_postgres_finance_ddl_uses_postgres_types(): statements: list[str] = [] @@ -110,6 +139,8 @@ async def test_finance_tables_are_created_idempotently_on_postgres(): with patch("backend.app.core.database.is_sqlite", return_value=False): await _migrate_create_finance_tables(conn) await _migrate_create_finance_tables(conn) + await _migrate_add_print_archive_cost_center(conn) + await _migrate_add_print_archive_cost_center(conn) await _migrate_create_finance_indexes(conn) await _migrate_create_finance_indexes(conn) @@ -127,8 +158,25 @@ async def test_finance_tables_are_created_idempotently_on_postgres(): "AND table_name = 'cost_centers' AND column_name = 'created_at'" ) ) + archive_cost_center = await conn.execute( + text( + "SELECT c.data_type, rc.delete_rule " + "FROM information_schema.columns c " + "JOIN information_schema.key_column_usage kcu " + " ON kcu.table_schema = c.table_schema " + " AND kcu.table_name = c.table_name " + " AND kcu.column_name = c.column_name " + "JOIN information_schema.referential_constraints rc " + " ON rc.constraint_schema = kcu.constraint_schema " + " AND rc.constraint_name = kcu.constraint_name " + "WHERE c.table_schema = 'public' " + "AND c.table_name = 'print_archives' " + "AND c.column_name = 'cost_center_id'" + ) + ) assert {row[0] for row in rows} == EXPECTED_TABLES assert timestamp_type.scalar_one() == "timestamp without time zone" + assert archive_cost_center.one() == ("integer", "SET NULL") finally: await engine.dispose() diff --git a/backend/tests/unit/test_printer_kill_switch.py b/backend/tests/unit/test_printer_kill_switch.py index 86812dfa9..9e6ff2202 100644 --- a/backend/tests/unit/test_printer_kill_switch.py +++ b/backend/tests/unit/test_printer_kill_switch.py @@ -1,4 +1,5 @@ from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest @@ -8,14 +9,21 @@ 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._kill_switch_notification_tasks.clear() main_module._expected_prints.clear() main_module._active_prints.clear() main_module._expected_print_registered_at.clear() + main_module._printer_reconciled_since_connect.clear() yield + for task in main_module._kill_switch_notification_tasks.values(): + if not task.done(): + task.cancel() main_module._unauthorized_print_kill_sent.clear() + main_module._kill_switch_notification_tasks.clear() main_module._expected_prints.clear() main_module._active_prints.clear() main_module._expected_print_registered_at.clear() + main_module._printer_reconciled_since_connect.clear() def test_gcode_3mf_status_filename_matches_registered_expected_print(): @@ -34,6 +42,8 @@ def test_gcode_3mf_status_filename_matches_registered_expected_print(): @pytest.mark.asyncio async def test_unauthorized_active_print_triggers_stop(monkeypatch): stop_calls: list[int] = [] + broadcast = AsyncMock() + provider_notification = AsyncMock(return_value=True) async def fake_status(*args, **kwargs): return None @@ -41,6 +51,9 @@ async def test_unauthorized_active_print_triggers_stop(monkeypatch): async def kill_switch_enabled(_db): return True + async def unauthorized(*_args): + return False + 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 @@ -50,6 +63,9 @@ async def test_unauthorized_active_print_triggers_stop(monkeypatch): monkeypatch.setattr(main_module, "printer_state_to_dict", lambda *args, **kwargs: {}) monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status) monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status) + monkeypatch.setattr(main_module.ws_manager, "broadcast", broadcast) + monkeypatch.setattr(main_module, "_is_bambuddy_authorized_print", unauthorized) + monkeypatch.setattr(main_module, "_send_kill_switch_provider_notification", provider_notification) monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled) state = SimpleNamespace( @@ -71,6 +87,7 @@ async def test_unauthorized_active_print_triggers_stop(monkeypatch): ams_filament_backup=False, current_print=None, subtask_name="foreign_job", + subtask_id="external-task-1", gcode_file="foreign_job.gcode", ) @@ -78,6 +95,39 @@ async def test_unauthorized_active_print_triggers_stop(monkeypatch): assert stop_calls == [7] assert 7 in main_module._unauthorized_print_kill_sent + broadcast.assert_awaited_once_with( + { + "type": "kill_switch_triggered", + "printer_id": 7, + "printer_name": "Printer 7", + "filename": "foreign_job", + "reason": "unauthorized_print", + } + ) + notification_task = main_module._kill_switch_notification_tasks[7] + assert await notification_task is True + provider_notification.assert_awaited_once_with( + 7, + "Printer 7", + { + "status": "stopped", + "filename": "foreign_job.gcode", + "subtask_name": "foreign_job", + "progress": 0, + "reason": "unauthorized_print", + }, + ) + + +@pytest.mark.asyncio +async def test_failed_immediate_notification_allows_completion_retry(): + task = main_module.spawn_background_task(_return_false(), name="test-kill-switch-notification-failure") + + assert await main_module._kill_switch_notification_already_sent(task) is False + + +async def _return_false(): + return False @pytest.mark.asyncio @@ -141,6 +191,9 @@ async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch): async def kill_switch_enabled(_db): return True + async def unauthorized(*_args): + return False + 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 @@ -150,6 +203,7 @@ async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch): monkeypatch.setattr(main_module, "printer_state_to_dict", lambda *args, **kwargs: {}) monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status) monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status) + monkeypatch.setattr(main_module, "_is_bambuddy_authorized_print", unauthorized) monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled) active_state = SimpleNamespace( @@ -171,6 +225,7 @@ async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch): ams_filament_backup=False, current_print=None, subtask_name="foreign_job", + subtask_id="external-task-1", gcode_file="foreign_job.gcode", ) @@ -193,6 +248,7 @@ async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch): ams_filament_backup=False, current_print=None, subtask_name="", + subtask_id=None, gcode_file=None, ) @@ -203,3 +259,90 @@ async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch): await main_module.on_printer_status_change(7, idle_state) assert 7 not in main_module._unauthorized_print_kill_sent + + +@pytest.mark.asyncio +@pytest.mark.parametrize("printer_state", ["RUNNING", "PAUSE"]) +async def test_persisted_print_is_authorized_after_restart(monkeypatch, printer_state): + archive = SimpleNamespace(id=123, filename="owned_job.gcode.3mf") + query_result = SimpleNamespace(scalar_one_or_none=lambda: archive) + db = SimpleNamespace(execute=AsyncMock(return_value=query_result)) + + class FakeSessionContext: + async def __aenter__(self): + return db + + async def __aexit__(self, *_args): + return False + + stop_calls: list[int] = [] + + async def fake_status(*args, **kwargs): + return None + + async def kill_switch_enabled(_db): + return True + + def discard_background_task(coro, **_kwargs): + coro.close() + + monkeypatch.setattr(main_module, "async_session", FakeSessionContext) + monkeypatch.setattr(main_module, "spawn_background_task", discard_background_task) + 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 *args, **kwargs: {}) + 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=printer_state, + progress=42, + remaining_time=600, + layer_num=50, + 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, + ams_filament_backup=False, + current_print=None, + subtask_name="owned_job", + subtask_id="bambuddy-task-123", + gcode_file="owned_job.gcode.3mf", + ) + + await main_module.on_printer_status_change(7, state) + + assert stop_calls == [] + assert (7, "owned_job.gcode.3mf") in main_module._active_prints + assert main_module._active_prints[(7, "owned_job.gcode.3mf")] == 123 + assert 7 not in main_module._unauthorized_print_kill_sent + + +@pytest.mark.asyncio +async def test_kill_switch_defers_when_restart_identity_is_not_available(monkeypatch): + state = SimpleNamespace( + current_print=None, + subtask_name="owned_job", + subtask_id=None, + gcode_file="owned_job.gcode.3mf", + ) + db = SimpleNamespace(execute=AsyncMock()) + + monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None) + + authorization = await main_module._is_bambuddy_authorized_print(7, state, db) + + assert authorization is None + db.execute.assert_not_awaited() diff --git a/backend/tests/unit/test_timelapse_baseline_restart_recovery.py b/backend/tests/unit/test_timelapse_baseline_restart_recovery.py index 3ab17cc81..ec49d91e4 100644 --- a/backend/tests/unit/test_timelapse_baseline_restart_recovery.py +++ b/backend/tests/unit/test_timelapse_baseline_restart_recovery.py @@ -98,7 +98,8 @@ async def test_running_observed_skips_when_baseline_already_present(): """If on_print_start already ran in this Bambuddy process for the same printer (the realistic same-session race), a second capture would overwrite the correct pre-print baseline with one taken later — which - could include the in-flight MP4. Skip when a baseline exists.""" + could include the in-flight MP4. The archive lookup still has to run so + restart recovery can restore durable print ownership for the kill switch.""" _timelapse_baselines[1] = {"pre_existing_a.mp4", "pre_existing_b.mp4"} with ( @@ -118,8 +119,9 @@ async def test_running_observed_skips_when_baseline_already_present(): }, ) - # Neither the DB lookup nor the FTP scan should have run. - mock_session_maker.assert_not_called() + # Ownership reconciliation still consults the DB, but the expensive + # timelapse scan remains one-shot. + mock_session_maker.assert_called_once() mock_list.assert_not_called() # Original baseline preserved. diff --git a/frontend/src/__tests__/hooks/useWebSocket.test.ts b/frontend/src/__tests__/hooks/useWebSocket.test.ts index cd9df397a..70a506192 100644 --- a/frontend/src/__tests__/hooks/useWebSocket.test.ts +++ b/frontend/src/__tests__/hooks/useWebSocket.test.ts @@ -6,7 +6,7 @@ */ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; -import { renderHook, waitFor, act } from '@testing-library/react'; +import { renderHook, waitFor, act, screen } from '@testing-library/react'; import React from 'react'; import { QueryClient, QueryClientProvider } from '@tanstack/react-query'; import { ToastProvider } from '../../contexts/ToastContext'; @@ -23,6 +23,10 @@ vi.mock('react-i18next', () => ({ const { printer, slots } = options as { printer: string; slots: string }; return `Missing assignments for ${printer}: ${slots}`; } + if (key === 'printers.toast.killSwitchTriggered' && options) { + const { printer, filename } = options as { printer: string; filename: string }; + return `Billing kill switch stopped ${filename} on ${printer}`; + } return key; }, i18n: {}, @@ -490,6 +494,28 @@ describe('useWebSocket hook', () => { vi.unstubAllGlobals(); }); + it('shows an error toast when the billing kill switch stops a print', async () => { + const { useWebSocket } = await import('../../hooks/useWebSocket'); + + renderHook(() => useWebSocket(), { + wrapper: createWrapper(queryClient), + }); + + const ws = await waitForWs(); + act(() => { + ws.open(); + ws.simulateMessage({ + type: 'kill_switch_triggered', + printer_id: 7, + printer_name: 'Printer B', + filename: 'foreign_job.3mf', + }); + }); + + const toast = screen.getByText('Billing kill switch stopped foreign_job.3mf on Printer B'); + expect(toast.parentElement).toHaveClass('bg-red-500/10'); + }); + it('handles spool_assignment_verified messages (success and failure) without error', async () => { const { useWebSocket } = await import('../../hooks/useWebSocket'); diff --git a/frontend/src/hooks/useWebSocket.ts b/frontend/src/hooks/useWebSocket.ts index 9b5ce77fc..feab6fb9e 100644 --- a/frontend/src/hooks/useWebSocket.ts +++ b/frontend/src/hooks/useWebSocket.ts @@ -17,6 +17,7 @@ interface WebSocketMessage { printer_id?: number; data?: Record; printer_name?: string; + filename?: string; missing_slots?: Array<{ slot?: string }>; // Spool-assignment read-back verification (#2582). slot?: string; @@ -338,6 +339,13 @@ export function useWebSocket() { debouncedInvalidate('archiveStats'); break; + case 'kill_switch_triggered': { + const printer = message.printer_name || `Printer ${message.printer_id ?? '?'}`; + const filename = message.filename || t('common.unknown'); + showToast(t('printers.toast.killSwitchTriggered', { printer, filename }), 'error'); + break; + } + case 'archive_created': debouncedInvalidate('archives'); debouncedInvalidate('archiveStats'); diff --git a/frontend/src/i18n/locales/de.ts b/frontend/src/i18n/locales/de.ts index a1009ab79..a4da534ce 100644 --- a/frontend/src/i18n/locales/de.ts +++ b/frontend/src/i18n/locales/de.ts @@ -447,6 +447,7 @@ export default { toast: { printerDeleted: 'Drucker gelöscht', missingSpoolAssignment: 'Druck gestartet auf {{printer}}. Fehlende Spulenzuordnung für: {{slots}}', + killSwitchTriggered: 'Der Billing-Kill-Switch hat einen nicht autorisierten Druck auf {{printer}} gestoppt: {{filename}}', assignmentVerified: 'Filament in Slot {{slot}} geladen ({{printer}})', assignmentVerifiedNoKprofile: 'Slot {{slot}} auf {{printer}} geladen, aber das Fluss-Kalibrierungsprofil (K-Profil) wurde nicht übernommen', assignmentNotConfirmed: 'Zuordnung für Slot {{slot}} auf {{printer}} konnte nicht bestätigt werden – bitte den AMS-Slot prüfen', diff --git a/frontend/src/i18n/locales/en.ts b/frontend/src/i18n/locales/en.ts index ac27b1aa1..a3297de62 100644 --- a/frontend/src/i18n/locales/en.ts +++ b/frontend/src/i18n/locales/en.ts @@ -450,6 +450,7 @@ export default { toast: { printerDeleted: 'Printer deleted', missingSpoolAssignment: 'Print started on {{printer}}. Missing spool assignment for: {{slots}}', + killSwitchTriggered: 'The billing kill switch stopped an unauthorized print on {{printer}}: {{filename}}', assignmentVerified: 'Filament loaded on slot {{slot}} ({{printer}})', assignmentVerifiedNoKprofile: 'Slot {{slot}} on {{printer}} loaded, but the flow calibration (K-profile) was not applied', assignmentNotConfirmed: 'Could not confirm the assignment for slot {{slot}} on {{printer}} — check the AMS slot', diff --git a/frontend/src/i18n/locales/es.ts b/frontend/src/i18n/locales/es.ts index 7f19f6769..6c46da64a 100644 --- a/frontend/src/i18n/locales/es.ts +++ b/frontend/src/i18n/locales/es.ts @@ -447,6 +447,7 @@ export default { toast: { printerDeleted: 'Impresora eliminada', missingSpoolAssignment: 'Impresión iniciada en {{printer}}. Falta la asignación de bobina para: {{slots}}', + killSwitchTriggered: 'El interruptor de seguridad de facturación detuvo una impresión no autorizada en {{printer}}: {{filename}}', assignmentVerified: 'Filamento cargado en la ranura {{slot}} ({{printer}})', assignmentVerifiedNoKprofile: 'Ranura {{slot}} en {{printer}} cargada, pero no se aplicó el perfil de calibración de flujo (perfil K)', assignmentNotConfirmed: 'No se pudo confirmar la asignación de la ranura {{slot}} en {{printer}}: revisa la ranura AMS', diff --git a/frontend/src/i18n/locales/fr.ts b/frontend/src/i18n/locales/fr.ts index eb4b1ad80..deb1f6572 100644 --- a/frontend/src/i18n/locales/fr.ts +++ b/frontend/src/i18n/locales/fr.ts @@ -447,6 +447,7 @@ export default { toast: { printerDeleted: 'Imprimante supprimée', missingSpoolAssignment: 'Impression démarrée sur {{printer}}. Attribution de bobine manquante pour : {{slots}}', + killSwitchTriggered: 'Le coupe-circuit de facturation a arrêté une impression non autorisée sur {{printer}} : {{filename}}', assignmentVerified: 'Filament chargé dans l\'emplacement {{slot}} ({{printer}})', assignmentVerifiedNoKprofile: 'Emplacement {{slot}} sur {{printer}} chargé, mais le profil de calibration de débit (profil K) n\'a pas été appliqué', assignmentNotConfirmed: 'Impossible de confirmer l\'attribution de l\'emplacement {{slot}} sur {{printer}} — vérifiez l\'emplacement AMS', diff --git a/frontend/src/i18n/locales/it.ts b/frontend/src/i18n/locales/it.ts index 8f033e42e..95991b497 100644 --- a/frontend/src/i18n/locales/it.ts +++ b/frontend/src/i18n/locales/it.ts @@ -447,6 +447,7 @@ export default { toast: { printerDeleted: 'Stampante eliminata', missingSpoolAssignment: 'Stampa avviata su {{printer}}. Mancano assegnazioni bobina per: {{slots}}', + killSwitchTriggered: 'L’interruttore di sicurezza della fatturazione ha arrestato una stampa non autorizzata su {{printer}}: {{filename}}', assignmentVerified: 'Filamento caricato nello slot {{slot}} ({{printer}})', assignmentVerifiedNoKprofile: 'Slot {{slot}} su {{printer}} caricato, ma il profilo di calibrazione del flusso (profilo K) non è stato applicato', assignmentNotConfirmed: 'Impossibile confermare l\'assegnazione dello slot {{slot}} su {{printer}} — controlla lo slot AMS', diff --git a/frontend/src/i18n/locales/ja.ts b/frontend/src/i18n/locales/ja.ts index 7f6d480d8..31b38ef89 100644 --- a/frontend/src/i18n/locales/ja.ts +++ b/frontend/src/i18n/locales/ja.ts @@ -446,6 +446,7 @@ export default { toast: { printerDeleted: 'プリンターを削除しました', missingSpoolAssignment: '{{printer}}で印刷を開始しました。以下のスプール割り当てがありません: {{slots}}', + killSwitchTriggered: '課金キルスイッチが{{printer}}で未承認の印刷を停止しました:{{filename}}', assignmentVerified: 'スロット{{slot}}にフィラメントを読み込みました({{printer}})', assignmentVerifiedNoKprofile: '{{printer}}のスロット{{slot}}を読み込みましたが、フロー校正プロファイル(Kプロファイル)は適用されませんでした', assignmentNotConfirmed: '{{printer}}のスロット{{slot}}の割り当てを確認できませんでした。AMSスロットを確認してください', diff --git a/frontend/src/i18n/locales/ko.ts b/frontend/src/i18n/locales/ko.ts index ef5b3db3b..cccac5722 100644 --- a/frontend/src/i18n/locales/ko.ts +++ b/frontend/src/i18n/locales/ko.ts @@ -421,6 +421,7 @@ export default { toast: { printerDeleted: '프린터가 삭제되었습니다', missingSpoolAssignment: '{{printer}}에서 인쇄가 시작되었습니다. 슬롯 할당 누락: {{slots}}', + killSwitchTriggered: '결제 킬 스위치가 {{printer}}에서 승인되지 않은 인쇄를 중지했습니다: {{filename}}', assignmentVerified: '슬롯 {{slot}}에 필라멘트가 로드되었습니다 ({{printer}})', assignmentVerifiedNoKprofile: '{{printer}}의 슬롯 {{slot}}이(가) 로드되었지만 유량 보정 프로파일(K 프로파일)이 적용되지 않았습니다', assignmentNotConfirmed: '{{printer}}의 슬롯 {{slot}} 할당을 확인할 수 없습니다. AMS 슬롯을 확인하세요', diff --git a/frontend/src/i18n/locales/pt-BR.ts b/frontend/src/i18n/locales/pt-BR.ts index a5ff4a5b8..d44fd0b14 100644 --- a/frontend/src/i18n/locales/pt-BR.ts +++ b/frontend/src/i18n/locales/pt-BR.ts @@ -447,6 +447,7 @@ export default { toast: { printerDeleted: 'Impressora excluída', missingSpoolAssignment: 'Impressão iniciada em {{printer}}. Atribuição de bobina ausente para: {{slots}}', + killSwitchTriggered: 'O bloqueio de segurança de cobrança interrompeu uma impressão não autorizada em {{printer}}: {{filename}}', assignmentVerified: 'Filamento carregado no compartimento {{slot}} ({{printer}})', assignmentVerifiedNoKprofile: 'Compartimento {{slot}} em {{printer}} carregado, mas o perfil de calibração de fluxo (perfil K) não foi aplicado', assignmentNotConfirmed: 'Não foi possível confirmar a atribuição do compartimento {{slot}} em {{printer}} — verifique o compartimento AMS', diff --git a/frontend/src/i18n/locales/ru.ts b/frontend/src/i18n/locales/ru.ts index 5cde0c590..74525762d 100644 --- a/frontend/src/i18n/locales/ru.ts +++ b/frontend/src/i18n/locales/ru.ts @@ -426,6 +426,7 @@ export default { toast: { printerDeleted: "Принтер удалён", missingSpoolAssignment: "На принтере {{printer}} началась печать. Не назначены катушки для слотов: {{slots}}", + killSwitchTriggered: 'Аварийный выключатель биллинга остановил несанкционированную печать на {{printer}}: {{filename}}', assignmentVerified: "Филамент загружен в слот {{slot}} ({{printer}})", assignmentVerifiedNoKprofile: "Слот {{slot}} на {{printer}} загружен, но профиль калибровки потока (K-профиль) не применён", assignmentNotConfirmed: "Не удалось подтвердить назначение слота {{slot}} на {{printer}} — проверьте слот AMS", diff --git a/frontend/src/i18n/locales/tr.ts b/frontend/src/i18n/locales/tr.ts index 9ab6996df..bbfe8a705 100644 --- a/frontend/src/i18n/locales/tr.ts +++ b/frontend/src/i18n/locales/tr.ts @@ -447,6 +447,7 @@ export default { toast: { printerDeleted: 'Yazıcı silindi', missingSpoolAssignment: '{{printer}} üzerinde baskı başladı. Şunlar için eksik makara ataması: {{slots}}', + killSwitchTriggered: 'Faturalandırma durdurma anahtarı {{printer}} üzerindeki yetkisiz baskıyı durdurdu: {{filename}}', assignmentVerified: '{{slot}} yuvasına filament yüklendi ({{printer}})', assignmentVerifiedNoKprofile: '{{printer}} üzerindeki {{slot}} yuvası yüklendi, ancak akış kalibrasyonu profili (K profili) uygulanmadı', assignmentNotConfirmed: '{{printer}} üzerindeki {{slot}} yuvası ataması doğrulanamadı — AMS yuvasını kontrol edin', diff --git a/frontend/src/i18n/locales/uk.ts b/frontend/src/i18n/locales/uk.ts index 3f6985d89..1cbe9fa5b 100644 --- a/frontend/src/i18n/locales/uk.ts +++ b/frontend/src/i18n/locales/uk.ts @@ -450,6 +450,7 @@ export default { toast: { printerDeleted: "Принтер видалено", missingSpoolAssignment: "Друк на {{printer}} розпочато. Для слотів {{slots}} не призначено котушки.", + killSwitchTriggered: 'Аварійний вимикач білінгу зупинив несанкціонований друк на {{printer}}: {{filename}}', assignmentVerified: "Філамент завантажено в слот AMS {{slot}} принтера {{printer}}", assignmentVerifiedNoKprofile: "Слот AMS {{slot}} на {{printer}} завантажено, але калібрування потоку (K-профіль) не застосовано", assignmentNotConfirmed: "Не вдалося підтвердити призначення для слота {{slot}} на {{printer}} — перевірте слот AMS", diff --git a/frontend/src/i18n/locales/zh-CN.ts b/frontend/src/i18n/locales/zh-CN.ts index b1432b986..3658b573a 100644 --- a/frontend/src/i18n/locales/zh-CN.ts +++ b/frontend/src/i18n/locales/zh-CN.ts @@ -447,6 +447,7 @@ export default { toast: { printerDeleted: '打印机已删除', missingSpoolAssignment: '已在{{printer}}上开始打印。以下料槽未分配耗材: {{slots}}', + killSwitchTriggered: '计费终止开关已停止 {{printer}} 上的未授权打印:{{filename}}', assignmentVerified: '耗材已加载到料槽{{slot}}({{printer}})', assignmentVerifiedNoKprofile: '{{printer}}的料槽{{slot}}已加载,但流量校准配置(K配置)未应用', assignmentNotConfirmed: '无法确认{{printer}}上料槽{{slot}}的分配,请检查AMS料槽', diff --git a/frontend/src/i18n/locales/zh-TW.ts b/frontend/src/i18n/locales/zh-TW.ts index 6809e716e..e0c014152 100644 --- a/frontend/src/i18n/locales/zh-TW.ts +++ b/frontend/src/i18n/locales/zh-TW.ts @@ -447,6 +447,7 @@ export default { toast: { printerDeleted: '印表機已刪除', missingSpoolAssignment: '已在{{printer}}上開始列印。以下料槽未分配耗材: {{slots}}', + killSwitchTriggered: '計費終止開關已停止 {{printer}} 上的未授權列印:{{filename}}', assignmentVerified: '耗材已載入料槽{{slot}}({{printer}})', assignmentVerifiedNoKprofile: '{{printer}}的料槽{{slot}}已載入,但流量校準設定檔(K設定檔)未套用', assignmentNotConfirmed: '無法確認{{printer}}上料槽{{slot}}的分配,請檢查AMS料槽',