"""Transfer-boundary day visibility — `access_until_date` is inclusive.

`PRODUCT_OWNERSHIP_ACCESS.access_until_date` is derived from
`dim_release_history.end_date_inclusive`: the *last* day of a former owner's
tenure, inclusive. Both layers honour that contract:

  * the dbt `transfer_time_slice` macro, which builds the `*_PRODUCT_TRANSFER`
    rollups, gates on `download_activity_date <= access_until_date`;
  * the ows-analytics v2 DAILY macros (`permissions_filter.sql`,
    `account_scope_filter.sql`, `account_product_filter.sql`) gate on
    `download_activity_date <= access_until_date`.

So a former owner sees streams *through and including* their transfer date.

These tests pin a single day — the cutoff day itself — and guard against a
regression to a strict `<` predicate, which would silently drop a former
owner's last day of streams.

Probes use /product/5244974/summary: product 5244974 has ~4,007 streams on
2026-04-30 (the originating owner's cutoff), so the signal is unambiguous.
"""

from __future__ import annotations

from datetime import date, timedelta

from analytics import config
from tests.integration.endpoints.conftest import PAGINATION, assert_endpoint
from tests.integration.transfer_ownership.conftest import (
    ALA_VENDOR,
    EMPLOYEE,
    ORIGINATING_BOUNDARY,
    PRODUCT_ID,
    PROFILE_PAIRS,
    SUED_VENDOR,
    assert_isolated_transfer,
)


def _single_day_total(headers, day: str) -> int:
    """TOTAL streams from /product/5244974/summary for a single-day window."""
    base = config.PRODUCT_SUMMARY_URL.replace("<product_id>", PRODUCT_ID)
    url = f"{base}?type=TOTAL&start_date={day}&end_date={day}&{PAGINATION}"
    payload = assert_endpoint(url, headers=headers)
    items = payload["items"]
    return int(items[0].get("streams") or 0) if items else 0


def _day_before(date_str: str) -> str:
    """Return the calendar day before an ISO date string."""
    y, m, d = (int(p) for p in date_str.split("-"))
    return (date(y, m, d) - timedelta(days=1)).isoformat()


def _day_after(date_str: str) -> str:
    """Return the calendar day after an ISO date string."""
    y, m, d = (int(p) for p in date_str.split("-"))
    return (date(y, m, d) + timedelta(days=1)).isoformat()


class TestOriginatingBoundary:
    """Boundary day for the originating owner: 2026-04-30."""

    def test_day_before_cutoff_is_visible(self):
        """Control: the former owner sees the day before its cutoff."""
        streams = _single_day_total(ALA_VENDOR, _day_before(ORIGINATING_BOUNDARY))
        assert streams > 0, (
            "former owner sees nothing the day before its cutoff — the probe "
            "has no data or the profile lacks access; fix that before reading "
            "the boundary tests"
        )

    def test_cutoff_day_is_visible_to_current_owner(self):
        """Control: the cutoff day genuinely has streams (current owner sees it)."""
        assert _single_day_total(SUED_VENDOR, ORIGINATING_BOUNDARY) > 0

    def test_cutoff_day_is_visible_to_former_owner(self):
        """Former owner sees its access_until_date day — the inclusive contract.

        Regression guard: a strict `<` predicate in the v2 DAILY macros would
        make this return zero.
        """
        assert _single_day_total(ALA_VENDOR, ORIGINATING_BOUNDARY) > 0

    def test_day_after_cutoff_is_visible_to_current_owner(self):
        """Control: 2026-05-01 has streams (the start of W2, visible to sued)."""
        assert _single_day_total(SUED_VENDOR, _day_after(ORIGINATING_BOUNDARY)) > 0

    def test_day_after_cutoff_is_invisible_to_former_owner(self):
        """Former owner sees nothing on the day AFTER its cutoff — the upper
        bound of the v2 truncation is inclusive at access_until_date, so day+1
        must drop to zero.

        Regression guard: a fuzzy `<= cutoff + 1` predicate (or a timezone
        bug pushing the boundary forward by a day) would let one extra day of
        post-transfer streams leak. Paired with test_cutoff_day_is_visible_
        to_former_owner this pins the boundary from both sides.
        """
        day_after = _day_after(ORIGINATING_BOUNDARY)
        streams = _single_day_total(ALA_VENDOR, day_after)
        assert streams == 0, (
            f"former owner leaked {streams} streams on day-after-cutoff "
            f"({day_after}) — v2 macro upper bound is wrong"
        )


class TestBoundaryEmployee:
    """The full-access employee sees the transfer-boundary day."""

    def test_employee_sees_boundary_day(self):
        """Employee (full access) sees streams on the transfer-boundary day."""
        assert _single_day_total(EMPLOYEE, ORIGINATING_BOUNDARY) > 0


class TestBoundaryFFPair:
    """Boundary-day FF-pair differential for the former owner.

    On 2026-04-30 (ala_vendor's cutoff) product 5244974 has ~4,007 streams.
    Legacy current-ownership (FF-OFF) hides the transferred-away product from
    ala entirely — so single-day total is 0. v2 (FF-ON) grants the historical
    slice and access_until_date is inclusive — so single-day total is ~4,007.

    Pins both behaviours in a single differential: the smallest, lowest-noise
    FF-pair test in the suite, because the window is one day with known
    traffic and no late-arriving-write drift.
    """

    def test_former_owner_boundary_day_grant(self):
        """FF-OFF: ala sees 0 on the cutoff day; FF-ON: ala sees its streams."""
        pair = PROFILE_PAIRS["ala_vendor"]
        on_total = _single_day_total(pair.ff_on, ORIGINATING_BOUNDARY)
        off_total = _single_day_total(pair.ff_off, ORIGINATING_BOUNDARY)
        assert_isolated_transfer(
            on_total,
            off_total,
            role_key=pair.key,
            label=(
                f"{pair.key} /product/{PRODUCT_ID}/summary day={ORIGINATING_BOUNDARY}"
            ),
        )