"""T-INV-001…024 (06-test-plan §3.6) against REAL migrated data in the
commit-safe m6 scratch clone. Unit-style tests run in the SAVEPOINT
session `s` — every service commit() rolls back with the test."""

import uuid
from datetime import datetime
from decimal import Decimal

import pytest
from sqlalchemy import func, select

from app.models import Invoice, InvoiceLine, Tenant, TimeEntry
from app.services import audit as audit_svc
from app.services.bulk import BulkSelection, bulk_delete, bulk_recalculate
from app.services.entries_write import require_can_edit, update_entry
from app.services.errors import ServiceError
from app.services.money import invoices as inv
from app.services.money.paste_block import paste_block_for_invoice
from app.services.reporting.entries import entry_totals
from app.services.reporting.filters import EntryFilters, EntryJoins, status_condition
from tests.invoices.conftest import pending_ids


def create(s, actor, name, **overrides) -> dict:
    payload = {"name": name, **overrides}
    return inv.create_invoice(s, actor, payload)


def apply_ids(s, actor, invoice_id, ids):
    return inv.apply_entries(
        s,
        actor,
        uuid.UUID(invoice_id) if isinstance(invoice_id, str) else invoice_id,
        BulkSelection(expected_count=len(ids), ids=list(ids)),
    )


def get_invoice(s, invoice_id) -> Invoice:
    return s.get(
        Invoice, uuid.UUID(invoice_id) if isinstance(invoice_id, str) else invoice_id
    )


def real_invoice_id(s, name: str) -> uuid.UUID:
    return s.scalars(select(Invoice.id).where(Invoice.name == name)).one()


# ------------------------------------------------------------- creation


def test_create_defaults_backdating_and_identity(s, rian):
    """B3/B4/B8/B9 + T-INV-011: status coerces, date defaults to the
    last day of the previous month, created_by is the REAL human, and
    creating straight to `paid` stamps BOTH timestamps."""
    row = create(s, rian, "M6 create test", status=" PAID ")
    invoice = get_invoice(s, row["id"])
    assert invoice.status == "paid"
    assert invoice.sent_at is not None and invoice.paid_at is not None
    assert invoice.invoice_date == inv.default_invoice_date()
    assert invoice.created_by == rian.real_user_id
    assert invoice.currency == "CAD"
    tenant_party = s.scalar(select(Tenant.party_id).where(Tenant.id == rian.tenant_id))
    assert invoice.from_party_id == tenant_party

    weird = create(s, rian, "M6 weird status", status="draft")
    assert weird["status"] == "open"  # coercion, not an error (B3)


def test_create_duplicate_name_case_insensitive(s, rian):
    """T-INV-001 — per tenant, case-insensitively, incl. against the
    real migrated names."""
    create(s, rian, "M6 Dup Test")
    with pytest.raises(ServiceError) as exc:
        create(s, rian, "m6 dup TEST")
    assert exc.value.code == "DUPLICATE_INVOICE_NAME"
    with pytest.raises(ServiceError):
        create(s, rian, "2026 may PLUSROI")  # real migrated invoice


def test_create_name_required(s, rian):
    with pytest.raises(ServiceError) as exc:
        create(s, rian, "   ")
    assert exc.value.code == "NAME_REQUIRED"
    assert exc.value.summary == "Invoice name is required."


def test_create_malformed_date_gets_default(s, rian):
    """T-INV-012 (create half): missing/malformed -> default."""
    row = create(s, rian, "M6 bad date", invoice_date="2026-99-99")
    assert row["invoice_date"] == inv.default_invoice_date().isoformat()


# --------------------------------------------------------------- status


def test_status_stamps_first_crossing_only(s, rian):
    """T-INV-004: open->sent stamps sent_at; ->paid stamps paid_at;
    paid->open->paid preserves the ORIGINAL timestamps."""
    row = create(s, rian, "M6 stamps")
    iid = uuid.UUID(row["id"])
    inv.update_invoice(s, rian, iid, {"status": "sent"})
    invoice = get_invoice(s, iid)
    first_sent = invoice.sent_at
    assert first_sent is not None and invoice.paid_at is None

    inv.update_invoice(s, rian, iid, {"status": "paid"})
    invoice = get_invoice(s, iid)
    first_paid = invoice.paid_at
    assert first_paid is not None

    inv.update_invoice(s, rian, iid, {"status": "open"})
    inv.update_invoice(s, rian, iid, {"status": "paid"})
    invoice = get_invoice(s, iid)
    assert invoice.sent_at == first_sent  # never cleared or overwritten
    assert invoice.paid_at == first_paid


def test_update_empty_date_writes_null(s, rian):
    """T-INV-012 (update half) — B66 parity."""
    row = create(s, rian, "M6 date null", invoice_date="2026-05-31")
    iid = uuid.UUID(row["id"])
    updated = inv.update_invoice(s, rian, iid, {"invoice_date": ""})
    assert updated["invoice_date"] is None


# ---------------------------------------------------------------- apply


def test_apply_stamps_lines_and_snapshots(s, rian):
    """T-INV-002: line linkage + attached_at/attached_by (= REAL user)
    + snapshot display fields as of attach (judgment #9)."""
    ids = pending_ids(s, 3)
    row = create(s, rian, "M6 apply")
    result = apply_ids(s, rian, row["id"], ids)
    assert result["applied"] == 3
    assert result["matched"] == 3
    assert sorted(result["applied_ids"]) == sorted(str(i) for i in ids)

    joins = EntryJoins()
    expected_names = {
        r[0]: (r[1], r[2])
        for r in s.execute(
            joins.select_from(
                TimeEntry.id, joins.client_expr, joins.project_expr
            ).where(TimeEntry.id.in_(ids))
        ).all()
    }
    for entry_id in ids:
        entry = s.get(TimeEntry, entry_id)
        assert entry.invoice_line_id is not None
        line = s.get(InvoiceLine, entry.invoice_line_id)
        assert line.invoice_id == uuid.UUID(row["id"])
        assert line.kind == "time"
        assert line.attached_by == rian.real_user_id
        assert line.attached_at is not None
        assert line.description == entry.description
        client, project = expected_names[entry_id]
        assert line.snapshot_client_name == client
        assert line.snapshot_project_name == project
        if entry.duration_seconds is not None:
            assert line.quantity == (
                Decimal(entry.duration_seconds) / Decimal(3600)
            ).quantize(Decimal("0.01"))
        assert line.amount == entry.billout_amount


def test_apply_skips_already_attached(s, rian):
    """T-INV-008: rows on invoice A are silently skipped when applying
    to invoice B; the verified count excludes them."""
    ids = pending_ids(s, 3)
    a = create(s, rian, "M6 skip A")
    apply_ids(s, rian, a["id"], ids[:2])
    b = create(s, rian, "M6 skip B")
    result = apply_ids(s, rian, b["id"], ids)
    assert result["applied"] == 1
    assert result["skipped_already_applied"] == 2
    # the two stayed on A
    for entry_id in ids[:2]:
        line = s.get(InvoiceLine, s.get(TimeEntry, entry_id).invoice_line_id)
        assert line.invoice_id == uuid.UUID(a["id"])


def test_apply_skips_blocked_and_agrees_with_display(s, rian):
    """T-INV-009/010: a row THE ONE predicate calls blocked is skipped
    by explicit-id apply — display status and apply eligibility agree
    by construction (the legacy operator-drift wart does not port)."""
    ids = pending_ids(s, 2)
    blocked_entry = s.get(TimeEntry, ids[0])
    blocked_entry.description = None  # now fails the one predicate
    s.flush()
    assert (
        s.scalar(
            select(func.count())
            .select_from(TimeEntry)
            .where(TimeEntry.id == ids[0], status_condition("blocked"))
        )
        == 1
    )
    row = create(s, rian, "M6 blocked skip")
    result = apply_ids(s, rian, row["id"], ids)
    assert result["applied"] == 1
    assert result["skipped_blocked"] == 1
    assert s.get(TimeEntry, ids[0]).invoice_line_id is None


def test_apply_currency_guard_aborts_before_write(s, rian):
    """Single-currency document rule (02 §3): a foreign-currency billout
    stamp aborts the WHOLE apply loudly; nothing is written."""
    ids = pending_ids(s, 2)
    odd = s.get(TimeEntry, ids[0])
    assert odd.billout_amount is not None
    odd.billout_currency = "USD"
    s.flush()
    row = create(s, rian, "M6 currency guard")
    with pytest.raises(ServiceError) as exc:
        apply_ids(s, rian, row["id"], ids)
    assert exc.value.code == "CURRENCY_MISMATCH"
    assert s.get(TimeEntry, ids[1]).invoice_line_id is None  # nothing applied


def test_apply_to_paid_invoice_refused(s, rian):
    """T-INV-020 (server side): a paid invoice takes no more entries;
    flipping back to sent restores it."""
    ids = pending_ids(s, 1)
    row = create(s, rian, "M6 paid target", status="paid")
    with pytest.raises(ServiceError) as exc:
        apply_ids(s, rian, row["id"], ids)
    assert exc.value.code == "INVOICE_PAID"
    inv.update_invoice(s, rian, uuid.UUID(row["id"]), {"status": "sent"})
    assert apply_ids(s, rian, row["id"], ids)["applied"] == 1


def test_apply_safety_stop_before_any_write(s, rian):
    """The M5 expected-count precondition carries to apply (T-ENT-17
    family): a stale count aborts with the exact message, no write."""
    ids = pending_ids(s, 2)
    row = create(s, rian, "M6 safety stop")
    with pytest.raises(ServiceError) as exc:
        inv.apply_entries(
            s,
            rian,
            uuid.UUID(row["id"]),
            BulkSelection(expected_count=3, ids=list(ids)),
        )
    assert exc.value.code == "SAFETY_STOP"
    assert exc.value.summary.startswith("Safety stop:")
    assert s.get(TimeEntry, ids[0]).invoice_line_id is None


def test_apply_all_matching_filter_verified_counts(s, rian):
    """T-INV-024 (the 343-vs-415 regression family): an all-matching
    apply attaches exactly the rows matching BOTH the display filter and
    eligibility; the verified count equals rows actually stamped."""
    filters = EntryFilters(
        status="pending",
        operator="PlusROI",
        start=datetime(2026, 6, 1),
        end=datetime(2026, 7, 1),
    )
    expected = entry_totals(s, rian, filters, zero_currency="CAD")["row_count"]
    assert expected > 0
    row = create(s, rian, "M6 all-matching")
    result = inv.apply_entries(
        s,
        rian,
        uuid.UUID(row["id"]),
        BulkSelection(expected_count=expected, filters=filters),
    )
    assert result["applied"] == expected
    stamped = s.scalar(
        select(func.count())
        .select_from(InvoiceLine)
        .where(InvoiceLine.invoice_id == uuid.UUID(row["id"]))
    )
    assert stamped == expected
    # and the filter now matches zero pending rows (they're applied)
    assert entry_totals(s, rian, filters, zero_currency="CAD")["row_count"] == 0


# --------------------------------------------------------------- detach


def test_detach_clears_lock_and_reports_honest_counts(s, rian):
    """T-INV-003 + T-INV-016: detach clears the linkage, deletes the
    line, and no-ops (counted) on rows that aren't attached."""
    ids = pending_ids(s, 3)
    row = create(s, rian, "M6 detach")
    apply_ids(s, rian, row["id"], ids[:2])
    line_ids = [s.get(TimeEntry, i).invoice_line_id for i in ids[:2]]

    result = inv.detach_entries(s, rian, ids=list(ids), expected_count=3)
    assert result["detached"] == 2
    assert result["skipped_unlocked"] == 1
    assert result["invoice_id"] == row["id"]
    for entry_id, line_id in zip(ids[:2], line_ids, strict=True):
        assert s.get(TimeEntry, entry_id).invoice_line_id is None
        assert s.get(InvoiceLine, line_id) is None  # line removed


def test_detach_scoped_to_one_invoice(s, rian):
    """The ?invoice= view's detach only touches THAT invoice's rows."""
    ids = pending_ids(s, 2)
    a = create(s, rian, "M6 detach A")
    b = create(s, rian, "M6 detach B")
    apply_ids(s, rian, a["id"], ids[:1])
    apply_ids(s, rian, b["id"], ids[1:])
    result = inv.detach_entries(
        s, rian, ids=list(ids), expected_count=2, invoice_id=uuid.UUID(a["id"])
    )
    assert result["detached"] == 1
    assert result["skipped_other_invoice"] == 1
    assert s.get(TimeEntry, ids[0]).invoice_line_id is None
    assert s.get(TimeEntry, ids[1]).invoice_line_id is not None


def test_detach_large_explicit_selection(s, rian):
    """T-INV-022: >150 ids in one set-based detach succeeds (the legacy
    unchunked .in() failure class does not port)."""
    ids = pending_ids(s, 160)
    assert len(ids) == 160
    row = create(s, rian, "M6 big detach")
    assert apply_ids(s, rian, row["id"], ids)["applied"] == 160
    result = inv.detach_entries(s, rian, ids=list(ids), expected_count=160)
    assert result["detached"] == 160
    assert (
        s.scalar(
            select(func.count())
            .select_from(InvoiceLine)
            .where(InvoiceLine.invoice_id == uuid.UUID(row["id"]))
        )
        == 0
    )


# ---------------------------------------------------------- permissions


def test_owner_gates_all_five_mutations(s, rian, adi):
    """T-INV-013: a manager is refused on every invoice mutation with
    the owner-only message and NO write happens."""
    row = create(s, rian, "M6 gates")
    iid = uuid.UUID(row["id"])
    ids = pending_ids(s, 1)

    with pytest.raises(ServiceError) as exc:
        inv.create_invoice(s, adi, {"name": "M6 adi invoice"})
    assert exc.value.summary == "Only an owner can create invoices."

    with pytest.raises(ServiceError) as exc:
        inv.update_invoice(s, adi, iid, {"name": "renamed"})
    assert exc.value.summary == "Only an owner can edit invoices."

    with pytest.raises(ServiceError) as exc:
        inv.delete_invoice(s, adi, iid)
    assert exc.value.summary == "Only an owner can delete invoices."

    with pytest.raises(ServiceError) as exc:
        apply_ids(s, adi, iid, ids)
    assert (
        exc.value.summary == "Only an organization owner can apply entries to invoices."
    )

    with pytest.raises(ServiceError) as exc:
        inv.detach_entries(s, adi, ids=list(ids), expected_count=1)
    assert exc.value.summary == "Only an owner can detach entries."

    assert get_invoice(s, iid).name == "M6 gates"
    assert s.get(TimeEntry, ids[0]).invoice_line_id is None


def test_locked_entry_guard_for_non_owner(s, rian, adi):
    """T-INV-014: an invoice-locked row refuses non-owner edits with the
    exact lock message (the write-side twin of the M5 read guard)."""
    ids = pending_ids(s, 1)
    row = create(s, rian, "M6 lock guard")
    apply_ids(s, rian, row["id"], ids)
    entry = s.get(TimeEntry, ids[0])
    with pytest.raises(ServiceError) as exc:
        require_can_edit(adi, entry, "edit")
    assert exc.value.status == 409
    assert exc.value.code == "ENTRY_LOCKED"
    assert (
        exc.value.summary
        == "This entry is attached to an invoice and locked. Only an owner can edit it."
    )


def test_bulk_ops_exclude_locked_rows(s, rian):
    """The M5 engine's lock rules hold against M6 locks: recalculate
    skips locked rows for everyone; bulk delete refuses an all-locked
    scope."""
    ids = pending_ids(s, 2)
    row = create(s, rian, "M6 bulk exclude")
    apply_ids(s, rian, row["id"], ids)

    recalc = bulk_recalculate(
        s, rian, BulkSelection(expected_count=2, ids=list(ids))
    )
    assert recalc["skipped_locked"] == 2
    assert recalc["changed"] == 0

    with pytest.raises(ServiceError) as exc:
        bulk_delete(s, rian, BulkSelection(expected_count=2, ids=list(ids)))
    assert exc.value.code == "ALL_LOCKED"


# --------------------------------------------------------------- delete


def test_delete_guarded_until_fully_detached(s, rian, adi):
    """T-INV-005 rebuilt (ADR #007): delete refuses while lines exist;
    after a full detach it deletes, and the entries are unlocked /
    editable by non-owners again."""
    ids = pending_ids(s, 2)
    row = create(s, rian, "M6 delete guard")
    iid = uuid.UUID(row["id"])
    apply_ids(s, rian, iid, ids)

    with pytest.raises(ServiceError) as exc:
        inv.delete_invoice(s, rian, iid)
    assert exc.value.code == "INVOICE_HAS_LINES"
    assert get_invoice(s, iid) is not None

    inv.detach_entries(s, rian, ids=list(ids), expected_count=2, invoice_id=iid)
    inv.delete_invoice(s, rian, iid)
    assert get_invoice(s, iid) is None
    entry = s.get(TimeEntry, ids[0])
    assert entry.invoice_line_id is None
    require_can_edit(adi, entry, "edit")  # unlocked -> manager may edit


# ------------------------------------------------------ totals / margin


def test_manual_total_override_changes_billout_and_margin(s, rian):
    """T-INV-006 + T-INV-021 (data half): the override replaces the
    displayed billout, margin follows, and the entry sum stays visible
    for the override note."""
    ids = pending_ids(s, 2)
    row = create(s, rian, "M6 override")
    iid = uuid.UUID(row["id"])
    apply_ids(s, rian, iid, ids)

    before = inv.invoice_detail(s, rian, iid)["invoice"]
    assert before["manual_override"] is False
    summed = Decimal(before["summed_billout"]["amount"])
    cost = Decimal(before["cost"]["amount"])
    assert Decimal(before["billout"]["amount"]) == summed

    inv.update_invoice(s, rian, iid, {"manual_total": "5000"})
    after = inv.invoice_detail(s, rian, iid)["invoice"]
    assert after["manual_override"] is True
    assert after["billout"]["amount"] == "5000.00"
    assert Decimal(after["summed_billout"]["amount"]) == summed
    assert Decimal(after["margin"]["amount"]) == Decimal("5000.00") - cost
    assert after["margin_pct"] is not None

    listed = {r["id"]: r for r in inv.list_invoices(s, rian)}[row["id"]]
    assert listed["billout"]["amount"] == "5000.00"

    # clearing the override restores the entry sum (B67)
    inv.update_invoice(s, rian, iid, {"manual_total": ""})
    cleared = inv.invoice_detail(s, rian, iid)["invoice"]
    assert Decimal(cleared["billout"]["amount"]) == summed


def test_breakdown_sums_equal_per_entry_sums(s, rian):
    """T-INV-007 on a REAL migrated invoice: breakdown groups equal the
    independent per-(client, project) sums over its attached entries."""
    iid = real_invoice_id(s, "2026 May PlusROI")
    breakdown = inv.invoice_breakdown(s, rian, iid)
    assert breakdown, "the real May invoice has lines"

    independent = {
        (r[0], r[1]): (r[2], Decimal(r[3] or 0), int(r[4]))
        for r in s.execute(
            select(
                InvoiceLine.snapshot_client_name,
                InvoiceLine.snapshot_project_name,
                func.count(TimeEntry.id),
                func.sum(TimeEntry.billout_amount),
                func.coalesce(func.sum(TimeEntry.duration_seconds), 0),
            )
            .select_from(TimeEntry)
            .join(InvoiceLine, InvoiceLine.id == TimeEntry.invoice_line_id)
            .where(InvoiceLine.invoice_id == iid)
            .group_by(
                InvoiceLine.snapshot_client_name, InvoiceLine.snapshot_project_name
            )
        ).all()
    }
    assert len(breakdown) == len(independent)
    for row in breakdown:
        count, billout, seconds = independent[(row["client"], row["project"])]
        assert row["row_count"] == count
        assert Decimal(row["summed_billout"]["amount"]) == billout.quantize(
            Decimal("0.01")
        )
        assert row["source_seconds"] == seconds
    # B44 sort: client case-insensitive, then project
    keys = [((r["client"] or "").lower(), r["project"] or "") for r in breakdown]
    assert keys == sorted(keys)

    detail = inv.invoice_detail(s, rian, iid)
    assert detail["invoice"]["entry_count"] == sum(
        r["row_count"] for r in breakdown
    )


def test_paste_block_ignores_manual_total(s, rian):
    """T-INV-017: amounts stay the per-row ENTRY sums (B51)."""
    iid = real_invoice_id(s, "2026 May PlusROI")
    before = paste_block_for_invoice(s, rian, iid)
    inv.update_invoice(s, rian, iid, {"manual_total": "99999"})
    after = paste_block_for_invoice(s, rian, iid)
    assert after["text"] == before["text"]


def test_paste_block_null_date_uses_default(s, rian):
    """T-INV-018: a NULL-dated invoice renders today's default in
    column 4; the list payload exposes created_at for the display
    fallback (B21)."""
    iid = real_invoice_id(s, "2026 April PlusROI")
    invoice = s.get(Invoice, iid)
    assert invoice.invoice_date is None
    block = paste_block_for_invoice(s, rian, iid)
    expected = inv.default_invoice_date().isoformat()
    assert block["invoice_date"] == expected
    first_line = block["text"].split("\n")[0].split("\t")
    assert first_line[3] == expected

    listed = {r["id"]: r for r in inv.list_invoices(s, rian)}[str(iid)]
    assert listed["invoice_date"] is None
    assert listed["created_at"] is not None


def test_invoice_filter_scopes_entries(s, rian):
    """T-INV-019: ?invoice= scopes the list AND totals to that invoice;
    the footer count matches the detail row_count."""
    iid = real_invoice_id(s, "2026 May PlusROI")
    totals = entry_totals(
        s, rian, EntryFilters(status="applied", invoice=iid), zero_currency="CAD"
    )
    detail = inv.invoice_detail(s, rian, iid)
    assert totals["row_count"] == detail["invoice"]["entry_count"]
    assert totals["row_count"] == detail["lines"]["total"]


def test_partial_visibility_sums_only_visible_rows(s, rian, adi):
    """T-INV-023: a manager's invoice sums cover only entries in their
    read scope (their own uploads) — never the whole invoice for free."""
    iid = real_invoice_id(s, "Legacy transfers")
    rian_detail = inv.invoice_detail(s, rian, iid)["invoice"]
    adi_detail = inv.invoice_detail(s, adi, iid)["invoice"]
    assert adi_detail["entry_count"] <= rian_detail["entry_count"]
    # the scoped count is exactly what adi's own entries query yields
    from app.services.authz.scope import scope

    adi_visible = s.scalar(
        scope(
            adi,
            "entries:read",
            select(func.count())
            .select_from(TimeEntry)
            .join(InvoiceLine, InvoiceLine.id == TimeEntry.invoice_line_id)
            .where(InvoiceLine.invoice_id == iid),
        )
    )
    assert adi_detail["entry_count"] == adi_visible


# ----------------------------------------------------------- audit/undo


def test_apply_detach_audited_but_not_generically_undoable(s, rian):
    """ADR #007: apply/detach write complete before-images but are NOT
    replayed by the generic undo — the inverse operation is the undo."""
    ids = pending_ids(s, 2)
    row = create(s, rian, "M6 audit policy")
    apply_ids(s, rian, row["id"], ids)

    from app.models import AuditLog

    apply_audit = s.scalars(
        select(AuditLog)
        .where(AuditLog.action == "invoices.apply_entries")
        .order_by(AuditLog.created_at.desc())
    ).first()
    assert apply_audit is not None
    assert len(apply_audit.before["rows"]) == 2  # complete before-images
    assert apply_audit.after["applied"] == 2

    with pytest.raises(ServiceError) as exc:
        audit_svc.undo_entries_action(s, rian, apply_audit.id)
    assert exc.value.code == "NOT_UNDOABLE"

    result = inv.detach_entries(s, rian, ids=list(ids), expected_count=2)
    detach_audit = s.scalars(
        select(AuditLog)
        .where(AuditLog.action == "invoices.detach_entries")
        .order_by(AuditLog.created_at.desc())
    ).first()
    assert detach_audit is not None
    assert len(detach_audit.before["lines"]) == 2  # line images for forensics
    assert result["detached"] == 2

    with pytest.raises(ServiceError) as exc:
        audit_svc.undo_entries_action(s, rian, detach_audit.id)
    assert exc.value.code == "NOT_UNDOABLE"


def test_update_entry_still_guarded_via_service(s, rian, adi):
    """Belt-and-braces on T-INV-014: the actual update_entry service
    refuses a manager on a locked row (not just the capability check).
    Uses a row adi can SEE (his own upload) so the guard — not scoping —
    is what refuses."""
    own = s.scalars(
        scope_own_pending(s, adi)
    ).first()
    if own is None:
        pytest.skip("no adi-visible pending row in the migrated data")
    row = create(s, rian, "M6 guard via service")
    apply_ids(s, rian, row["id"], [own.id])
    with pytest.raises(ServiceError) as exc:
        update_entry(s, adi, own.id, {"description": "nope"})
    assert exc.value.code == "ENTRY_LOCKED"


def scope_own_pending(s, actor):
    from app.services.authz.scope import scope

    return scope(
        actor,
        "entries:read",
        select(TimeEntry).where(status_condition("pending")).order_by(TimeEntry.id),
    )