"""The §1 filter contract over the migrated fixture graph: ILIKE
contains, NULL-column exclusion, the exact q column set, half-open date
windows, batch / imported_by / invoice / missing_info, and THE ONE
status predicate (partition property + transferred alias + unknown
status matches nothing). T-ENT-02/03/04/05/06 + filter-contract rows.
"""

import uuid

import pytest
from sqlalchemy import select

from app.services.reporting.entries import entry_totals, list_entries
from app.services.reporting.filters import EntryFilters
from tests.reporting.conftest import user_id

# fixture ids (tests/migrate/fixtures/legacy_fixture.sql; ids carry)
INV_MAY = uuid.UUID("0b000000-0000-0000-0000-000000000001")
IMP_CLOCKIFY = uuid.UUID("09000000-0000-0000-0000-000000000001")


def descriptions(session, actor, f: EntryFilters) -> set[str]:
    page = list_entries(session, actor, f)
    return {row["description"] for row in page.rows}


def count(session, actor, f: EntryFilters) -> int:
    return entry_totals(session, actor, f)["row_count"]


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

def test_t_ent_03_status_partition(rpt_session, rian):
    """pending + blocked + applied = all — every row in exactly one."""
    totals = {
        status: count(rpt_session, rian, EntryFilters(status=status))
        for status in ("all", "pending", "blocked", "applied")
    }
    assert totals["all"] == 7
    assert totals["pending"] + totals["blocked"] + totals["applied"] == totals["all"]
    assert totals["applied"] == 2  # te-1 (invoice) + te-6 (Legacy transfers)
    assert totals["blocked"] == 2  # te-4 (unresolved) + te-5 (no end_at)


def test_t_ent_02_transferred_is_an_alias_of_applied(rpt_session, rian):
    applied = descriptions(rpt_session, rian, EntryFilters(status="applied"))
    transferred = descriptions(rpt_session, rian, EntryFilters(status="transferred"))
    assert applied == transferred == {"March work A", "legacy admin work"}


def test_unknown_status_matches_nothing(rpt_session, rian):
    assert count(rpt_session, rian, EntryFilters(status="bogus")) == 0


def test_blocked_rows_and_reasons(rpt_session, rian):
    page = list_entries(rpt_session, rian, EntryFilters(status="blocked"))
    by_desc = {row["description"]: row for row in page.rows}
    # te-4: no worker resolved — first missing field wins
    assert by_desc["who did this"]["blocked_reason"] == (
        "Unknown person — pick a Team member."
    )
    # te-5: baseline sentinel is blocked only by its missing end time
    baseline = by_desc["Transfer baseline from Toggl, totals as of March 31, 2026"]
    assert baseline["blocked_reason"] == "Missing end time."


# ------------------------------------------------------------- text filters

def test_ilike_contains_case_insensitive(rpt_session, rian):
    """'roi' ~ 'PlusROI' (contains, case-insensitive) on the operator
    DISPLAY expression — resolved rows via the FK chain."""
    rows = descriptions(rpt_session, rian, EntryFilters(operator="roi"))
    assert "March work A" in rows and "feb toggl work" in rows
    # te-4 is unassigned with raw_operator='PlusROI' — matches via raw_*
    assert "who did this" in rows
    assert "automation run" not in rows  # Bowden Works operator


def test_null_column_exclusion(rpt_session, rian):
    """ILIKE on a NULL display column excludes the row: te-4 has no
    client (raw_client NULL) so ANY client filter drops it."""
    everything = descriptions(rpt_session, rian, EntryFilters())
    assert "who did this" in everything
    filtered = descriptions(rpt_session, rian, EntryFilters(client="e"))
    assert "who did this" not in filtered


def test_t_ent_05_q_searches_exactly_four_columns(rpt_session, rian):
    # matches via description only
    assert descriptions(rpt_session, rian, EntryFilters(q="who did")) == {"who did this"}
    # worker display name is NOT in the q column set...
    assert descriptions(rpt_session, rian, EntryFilters(q="Rian Bowden")) == set()
    # ...but the user= filter matches it (contains)
    assert "legacy admin work" in descriptions(
        rpt_session, rian, EntryFilters(user="rian bow")
    )
    # q matches source_user_email
    assert "automation run" in descriptions(rpt_session, rian, EntryFilters(q="gary@"))
    # q matches client display name
    assert "March work A" in descriptions(rpt_session, rian, EntryFilters(q="Brentwood"))


def test_wildcards_in_user_input_stay_live(rpt_session, rian):
    """Raw pattern concatenation (parity): '%' in input acts as a
    wildcard."""
    rows = descriptions(rpt_session, rian, EntryFilters(project="Site%build"))
    assert "March work A" in rows


# ---------------------------------------------------------------- windows

def test_t_ent_04_half_open_date_window(rpt_session, rian):
    from datetime import datetime

    f = EntryFilters(
        start=datetime(2026, 3, 1, 0, 0, 0), end=datetime(2026, 3, 2, 0, 0, 0)
    )
    assert descriptions(rpt_session, rian, f) == {"March work A"}
    # end is EXCLUSIVE: extending to 03-03 pulls in the 03-02 row
    f2 = EntryFilters(
        start=datetime(2026, 3, 1, 0, 0, 0), end=datetime(2026, 3, 3, 0, 0, 0)
    )
    assert descriptions(rpt_session, rian, f2) == {"March work A", "March work B"}


# ------------------------------------------------------- id-based filters

def test_batch_filter(rpt_session, rian):
    f = EntryFilters(batch=IMP_CLOCKIFY)
    assert count(rpt_session, rian, f) == 4  # te-1..te-4


def test_t_ent_06_imported_by(rpt_session, rian):
    adi_id = user_id(rpt_session, "adi")
    assert count(rpt_session, rian, EntryFilters(imported_by=adi_id)) == 4
    # a user with zero imports -> zero rows, no sentinel leakage
    assert count(rpt_session, rian, EntryFilters(imported_by=uuid.uuid4())) == 0


def test_t_ent_07_invoice_filter_scopes_list_AND_totals(rpt_session, rian):
    f = EntryFilters(invoice=INV_MAY)
    assert descriptions(rpt_session, rian, f) == {"March work A"}
    totals = entry_totals(rpt_session, rian, f)
    assert totals["row_count"] == 1
    assert totals["source_seconds"] == 3600
    assert totals["billout"]["amount"] == "35.00"


def test_missing_info(rpt_session, rian):
    rows = descriptions(rpt_session, rian, EntryFilters(missing_info=True))
    assert rows == {"who did this"}  # the only row with a display gap


# ------------------------------------------------------------ visibility

def test_totals_respect_scope_for_manager(rpt_session, adi):
    """Aggregates are scope()-filtered exactly like lists (rpcs §0
    SECURITY INVOKER note): adi's totals cover ONLY his batches."""
    totals = entry_totals(rpt_session, adi, EntryFilters())
    assert totals["row_count"] == 4
    assert "billout" not in totals  # T-PRM-05: absent, not nulled


def test_pagination_shape(rpt_session, rian):
    page = list_entries(rpt_session, rian, EntryFilters(), page=1)
    assert page.total_count == 7
    assert page.total_pages == 1
    assert page.page_size == 100
    # sorted date desc by default
    dates = [row["start_at"] for row in page.rows]
    assert dates == sorted(dates, reverse=True)


@pytest.mark.parametrize(
    "sort_key",
    ["date", "user", "operator", "client", "project", "description",
     "source_hrs", "cost", "billout_amount", "status"],
)
def test_t_ent_08_every_sort_key_works_nulls_last(rpt_session, rian, sort_key):
    page = list_entries(rpt_session, rian, EntryFilters(), sort=sort_key, direction="asc")
    assert page.total_count == 7 and len(page.rows) == 7
    if sort_key == "cost":
        costs = [row["cost"] for row in page.rows]
        nulls = [c is None for c in costs]
        assert nulls == sorted(nulls)  # NULLs last on asc
        amounts = [float(c["amount"]) for c in costs if c is not None]
        assert amounts == sorted(amounts)


def test_owner_totals_cover_every_tenant_row(rpt_session, rian):
    from sqlalchemy import func

    from app.models import TimeEntry

    total = rpt_session.scalar(select(func.count()).select_from(TimeEntry))
    assert count(rpt_session, rian, EntryFilters()) == total == 7