bambuddy/backend/tests/unit/test_print_log.py

"""Unit tests for print log service and schema."""

from datetime import datetime, timedelta
from unittest.mock import AsyncMock, MagicMock

import pytest

from backend.app.schemas.print_log import PrintLogEntrySchema, PrintLogResponse
from backend.app.services.print_log import write_log_entry


class TestPrintLogEntrySchema:
    """Test PrintLogEntrySchema validation."""

    def test_minimal_entry(self):
        """Schema accepts minimal required fields."""
        entry = PrintLogEntrySchema(
            id=1,
            status="completed",
            created_at=datetime(2024, 1, 15, 10, 30, 0),
        )
        assert entry.id == 1
        assert entry.status == "completed"
        assert entry.print_name is None
        assert entry.printer_name is None
        assert entry.duration_seconds is None

    def test_full_entry(self):
        """Schema accepts all fields."""
        started = datetime(2024, 1, 15, 10, 0, 0)
        completed = datetime(2024, 1, 15, 12, 30, 0)
        entry = PrintLogEntrySchema(
            id=42,
            print_name="Benchy",
            printer_name="X1C-01",
            printer_id=3,
            status="completed",
            started_at=started,
            completed_at=completed,
            duration_seconds=9000,
            filament_type="PLA",
            filament_color="#FF5500",
            filament_used_grams=15.2,
            thumbnail_path="archives/3/20240115_benchy/thumbnail.png",
            created_by_username="admin",
            created_at=datetime(2024, 1, 15, 12, 30, 0),
        )
        assert entry.print_name == "Benchy"
        assert entry.printer_name == "X1C-01"
        assert entry.filament_used_grams == 15.2
        assert entry.created_by_username == "admin"

    def test_failed_status(self):
        """Schema accepts various status values."""
        for status in ("completed", "failed", "stopped", "cancelled", "skipped"):
            entry = PrintLogEntrySchema(id=1, status=status, created_at=datetime.now())
            assert entry.status == status


class TestPrintLogResponse:
    """Test PrintLogResponse pagination wrapper."""

    def test_empty_response(self):
        """Empty response with zero total."""
        resp = PrintLogResponse(items=[], total=0)
        assert len(resp.items) == 0
        assert resp.total == 0

    def test_paginated_response(self):
        """Response with items and total count > items count."""
        items = [PrintLogEntrySchema(id=i, status="completed", created_at=datetime.now()) for i in range(3)]
        resp = PrintLogResponse(items=items, total=100)
        assert len(resp.items) == 3
        assert resp.total == 100


class TestWriteLogEntry:
    """Test the write_log_entry service function (logic only, no DB)."""

    def test_duration_calculation(self):
        """Duration is computed from started_at and completed_at."""
        started = datetime(2024, 1, 15, 10, 0, 0)
        completed = started + timedelta(hours=2, minutes=30)

        # Simulating the duration calculation from write_log_entry
        duration = int((completed - started).total_seconds())
        assert duration == 9000  # 2.5 hours = 9000 seconds

    def test_duration_none_when_missing_times(self):
        """Duration is None when started_at or completed_at is missing."""
        started = datetime(2024, 1, 15, 10, 0, 0)
        completed_at = None
        started_at = None
        completed = datetime.now()

        # No completed_at
        duration = None
        if started and completed_at:
            duration = int((completed_at - started).total_seconds())
        assert duration is None

        # No started_at
        duration = None
        if started_at and completed:
            duration = int((completed - started_at).total_seconds())
        assert duration is None


class TestWriteLogEntryReconciledDuration:
    """write_log_entry duration handling for reconciled (synthetic) completions (#2592).

    A reconciled abort closes out a stale ``status="printing"`` archive at
    reconnect; its real end time is unknown, so ``completed_at - started_at``
    would bank the whole disconnect gap as print time. Those entries must log
    0, while genuine prints (including >24h ones) keep their real duration.
    """

    @staticmethod
    async def _write(**kwargs):
        db = MagicMock()
        db.flush = AsyncMock()
        return await write_log_entry(db, **kwargs)

    @pytest.mark.asyncio
    async def test_reconciled_logs_zero_despite_multiday_gap(self):
        started = datetime(2026, 7, 15, 10, 0, 0)
        completed = started + timedelta(days=2, hours=4)  # the reconnect moment, not the real end
        entry = await self._write(status="aborted", started_at=started, completed_at=completed, reconciled=True)
        assert entry.duration_seconds == 0

    @pytest.mark.asyncio
    async def test_reconciled_logs_zero_even_without_timestamps(self):
        entry = await self._write(status="aborted", reconciled=True)
        assert entry.duration_seconds == 0

    @pytest.mark.asyncio
    async def test_genuine_long_print_retains_full_duration(self):
        """A legitimate >24h print keeps its real duration — no cap, no zeroing."""
        started = datetime(2026, 7, 15, 10, 0, 0)
        completed = started + timedelta(hours=30)
        entry = await self._write(status="completed", started_at=started, completed_at=completed)
        assert entry.duration_seconds == 30 * 3600

    @pytest.mark.asyncio
    async def test_non_reconciled_missing_times_is_none(self):
        entry = await self._write(status="completed", started_at=datetime(2026, 7, 15, 10, 0, 0))
        assert entry.duration_seconds is None


class TestSchemaValidatesFromOrmRow:
    """#2636: the Print Log's cost and energy columns read empty for every
    run because both routes built the response field-by-field and simply
    never mentioned ``cost`` / ``energy_kwh`` / ``energy_cost``. Pydantic
    filled the gap with each field's default, so a dropped field looked
    exactly like a NULL column on the wire — no error, no log line. The
    same trap had already eaten ``failure_reason`` once (#1687 part 4).

    Validating from the ORM row is what removes the chance to forget one, so
    these tests pin the mechanism rather than any particular field list.
    """

    @staticmethod
    def _row(**overrides):
        row = MagicMock()
        row.id = 7
        row.archive_id = 3
        row.print_name = "Benchy"
        row.printer_name = "X1C-01"
        row.printer_id = 1
        row.status = "completed"
        row.started_at = datetime(2026, 7, 24, 18, 35, 0)
        row.completed_at = datetime(2026, 7, 24, 19, 24, 0)
        row.duration_seconds = 2940
        row.filament_type = "PLA"
        row.filament_color = "#000000"
        row.filament_used_grams = 15.5
        row.cost = 0.42
        row.energy_kwh = 0.31
        row.energy_cost = 0.09
        # Non-null so `test_every_declared_field_is_carried` can assert that
        # nothing falls back to its default.
        row.failure_reason = "warping"
        row.user_verdict = "good"
        row.thumbnail_path = "archives/1/x/thumbnail.png"
        row.created_by_id = 2
        row.created_by_username = "martin"
        row.created_at = datetime(2026, 7, 24, 18, 35, 0)
        for k, v in overrides.items():
            setattr(row, k, v)
        return row

    def test_money_and_energy_survive_the_round_trip(self):
        entry = PrintLogEntrySchema.model_validate(self._row())
        assert entry.cost == 0.42
        assert entry.energy_kwh == 0.31
        assert entry.energy_cost == 0.09
        assert entry.filament_used_grams == 15.5

    def test_every_declared_field_is_carried(self):
        """Nothing on the schema may come back as its default when the row
        has a value — that is the whole failure mode, generalised."""
        entry = PrintLogEntrySchema.model_validate(self._row())
        for name in PrintLogEntrySchema.model_fields:
            assert getattr(entry, name) is not None, f"{name} was dropped in serialisation"

    def test_a_genuinely_null_column_stays_null(self):
        """The counterpart: energy is written by a background task after the
        row, so a just-finished print really has none. That must read as
        None, not as a fabricated zero."""
        entry = PrintLogEntrySchema.model_validate(self._row(energy_kwh=None, energy_cost=None))
        assert entry.energy_kwh is None
        assert entry.energy_cost is None
        assert entry.cost == 0.42