"""The ambient bubble: what it picks, and — mostly — what it refuses to.

Every assertion here is about rationing. The surface itself is three
lines of code; the reason it is worth having is that it does NOT show
everything to everyone every day, and each of those refusals is a
separate rule that can regress independently.
"""
from __future__ import annotations

from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select, update

from app.domains.security_awareness.models import SecawImpression
from tests.test_security_awareness.conftest import (
    ADMIN_BASE,
    SELF_BASE,
    article_payload,
    headers_for,
    make_person,
)

pytestmark = pytest.mark.asyncio

BUBBLE = f"{SELF_BASE}/bubble"
SETTINGS = f"{ADMIN_BASE}/bubble-settings"


async def _publish(client, headers, title: str) -> str:
    payload = article_payload()
    payload["localized"]["zh"]["title"] = title
    r = await client.post(
        f"{ADMIN_BASE}/materials", json=payload, headers=headers,
    )
    assert r.status_code == 201, r.text
    return r.json()["data"]["id"]


async def _set_caps(client, headers, **kw) -> None:
    r = await client.put(SETTINGS, json=kw, headers=headers)
    assert r.status_code == 200, r.text


async def _age_bubble_impressions(db, person_id, *, hours: int) -> None:
    """Push this person's bubble exposures back in time.

    Used to cross the 24-hour "already met today" window without
    sleeping. The dedupe window inside ``record_impression`` is separate
    and is exercised by the impressions suite.
    """
    await db.execute(
        update(SecawImpression)
        .where(
            SecawImpression.person_id == person_id,
            SecawImpression.surface == "bubble",
        )
        .values(occurred_at=datetime.now(UTC) - timedelta(hours=hours))
    )
    await db.commit()


# ── The daily-new cap ───────────────────────────────────────────────


async def test_one_new_material_per_day_by_default(
    client, security_officer, reader,
):
    """An ambient surface earns its keep by repeating one idea, not by
    volume."""
    officer, oh = security_officer
    _person, rh = reader
    for i in range(4):
        await _publish(client, oh, f"素材 {i}")

    data = (await client.get(BUBBLE, headers=rh)).json()["data"]
    assert data["enabled"] is True
    assert len(data["items"]) == 1


async def test_the_pick_is_stable_within_the_day(
    client, security_officer, reader,
):
    """Recomputing on every page load would make the bubble reshuffle as
    somebody navigates, which reads as noise rather than as a message."""
    officer, oh = security_officer
    _person, rh = reader
    for i in range(5):
        await _publish(client, oh, f"素材 {i}")

    picks = {
        (await client.get(BUBBLE, headers=rh)).json()["data"]["items"][0][
            "material_id"
        ]
        for _ in range(5)
    }
    assert len(picks) == 1


async def test_different_people_are_not_all_shown_the_same_card(
    client, db_session, security_officer, reader,
):
    """The whole company staring at the same card on the same morning is
    an ambient surface working against itself."""
    officer, oh = security_officer
    for i in range(12):
        await _publish(client, oh, f"素材 {i}")

    picks: set[str] = set()
    for n in range(8):
        person = await make_person(db_session, f"r{n}@test.com", role="employee")
        await db_session.commit()
        data = (await client.get(
            BUBBLE, headers=headers_for(person, ["employee"]),
        )).json()["data"]
        picks.add(data["items"][0]["material_id"])
    assert len(picks) > 1, "every person got the identical card"


async def test_raising_the_cap_widens_the_rotation(
    client, security_officer, reader,
):
    officer, oh = security_officer
    _person, rh = reader
    for i in range(4):
        await _publish(client, oh, f"素材 {i}")

    await _set_caps(client, oh, daily_new_materials=3)
    data = (await client.get(BUBBLE, headers=rh)).json()["data"]
    assert len(data["items"]) == 3


async def test_a_cap_of_zero_pauses_new_material_without_blanking_today(
    client, db_session, security_officer, reader,
):
    """Zero is a legitimate way to pause the surface. What is already in
    today's rotation stays on screen — blanking the bubble mid-day would
    look like a fault, not like a decision."""
    officer, oh = security_officer
    person, rh = reader
    await _publish(client, oh, "今天这条")

    first = (await client.get(BUBBLE, headers=rh)).json()["data"]["items"]
    assert len(first) == 1
    shown = first[0]["material_id"]
    # The client reports what it rendered — that is what makes it
    # "already met today".
    await client.post(
        f"{SELF_BASE}/impressions",
        json={"material_id": shown, "surface": "bubble"}, headers=rh,
    )

    await _set_caps(client, oh, daily_new_materials=0)
    still = (await client.get(BUBBLE, headers=rh)).json()["data"]["items"]
    assert [i["material_id"] for i in still] == [shown]

    # …but a person who has met nothing today now gets nothing.
    fresh = await make_person(db_session, "fresh@test.com", role="employee")
    await db_session.commit()
    theirs = (await client.get(
        BUBBLE, headers=headers_for(fresh, ["employee"]),
    )).json()["data"]
    assert theirs["items"] == []


# ── The repeat cap ──────────────────────────────────────────────────


async def test_a_material_stops_appearing_after_the_repeat_cap(
    client, db_session, security_officer, reader,
):
    """After three showings it has either worked or it has not; a fourth
    only teaches the person to ignore the bubble."""
    officer, oh = security_officer
    person, rh = reader
    only = await _publish(client, oh, "唯一一条")

    for day in range(3):
        items = (await client.get(BUBBLE, headers=rh)).json()["data"]["items"]
        assert [i["material_id"] for i in items] == [only], f"day {day}"
        await client.post(
            f"{SELF_BASE}/impressions",
            json={"material_id": only, "surface": "bubble"}, headers=rh,
        )
        # Age everything so the next loop is a new "day" AND outside the
        # 30-minute dedupe window.
        await _age_bubble_impressions(db_session, person.id, hours=30 * (day + 1))

    assert (await client.get(BUBBLE, headers=rh)).json()["data"]["items"] == []


async def test_opening_a_material_takes_it_out_of_the_bubble(
    client, security_officer, reader,
):
    """They did the thing. Still nudging them about it is the fastest way
    to make the nudge worthless."""
    officer, oh = security_officer
    _person, rh = reader
    only = await _publish(client, oh, "读过就不再推")

    assert (await client.get(
        BUBBLE, headers=rh,
    )).json()["data"]["items"][0]["material_id"] == only

    await client.get(f"{SELF_BASE}/materials/{only}", headers=rh)

    assert (await client.get(BUBBLE, headers=rh)).json()["data"]["items"] == []


# ── Eligibility ─────────────────────────────────────────────────────


async def test_only_published_material_reaches_the_bubble(
    client, security_officer, reader,
):
    officer, oh = security_officer
    _person, rh = reader
    r = await client.post(
        f"{ADMIN_BASE}/materials", json=article_payload(publish=False),
        headers=oh,
    )
    assert r.status_code == 201
    assert (await client.get(BUBBLE, headers=rh)).json()["data"]["items"] == []


async def test_the_feed_records_nothing_by_itself(
    client, db_session, security_officer, reader,
):
    """A browser receiving a rotation list is not evidence that any of it
    reached anybody's eyes. The client reports what it RENDERED — which
    is the opposite of the material-detail route, where a human opening
    the page IS the event."""
    officer, oh = security_officer
    person, rh = reader
    await _publish(client, oh, "只取不记")

    for _ in range(3):
        await client.get(BUBBLE, headers=rh)

    rows = (await db_session.execute(
        select(SecawImpression).where(SecawImpression.person_id == person.id)
    )).scalars().all()
    assert rows == []


async def test_switching_the_surface_off_says_so_rather_than_going_empty(
    client, security_officer, reader,
):
    """``enabled: false`` and "you are up to date" are different states,
    and the client renders them differently (tips fallback vs. nothing
    new) — collapsing both into an empty list would hide an admin's
    decision behind a normal-looking response."""
    officer, oh = security_officer
    _person, rh = reader
    await _publish(client, oh, "被关掉了")

    await _set_caps(client, oh, enabled=False)
    data = (await client.get(BUBBLE, headers=rh)).json()["data"]
    assert data["enabled"] is False
    assert data["items"] == []


async def test_the_hook_is_picked_server_side(
    client, security_officer, reader,
):
    """The bubble is one line next to a thumbnail. The service picks
    summary-or-title rather than letting CSS truncate a 500-character
    summary into nonsense."""
    officer, oh = security_officer
    _person, rh = reader
    payload = article_payload()
    payload["localized"]["zh"]["summary"] = "一句话摘要"
    r = await client.post(
        f"{ADMIN_BASE}/materials", json=payload, headers=oh,
    )
    assert r.status_code == 201

    item = (await client.get(BUBBLE, headers=rh)).json()["data"]["items"][0]
    assert item["hook"] == "一句话摘要"

    # With no summary the title is the honest fallback, not an empty line.
    second = await _publish(client, oh, "没有摘要")
    await _set_caps(client, oh, daily_new_materials=5)
    items = (await client.get(BUBBLE, headers=rh)).json()["data"]["items"]
    hooks = {i["material_id"]: i["hook"] for i in items}
    assert hooks[second] == "没有摘要"


# ── Settings ────────────────────────────────────────────────────────


async def test_a_malformed_setting_falls_back_instead_of_breaking(
    client, db_session, security_officer, reader,
):
    """The bubble hangs off every authenticated page. A bad value in
    ``platform_settings`` — which the generic settings page can also
    write — must not surface as a broken app shell."""
    from app.integrations.models import PlatformSetting

    officer, oh = security_officer
    _person, rh = reader
    await _publish(client, oh, "坏配置也要能读")

    db_session.add(PlatformSetting(
        key="security_awareness.bubble.daily_new_materials",
        value="not-a-number",
    ))
    await db_session.commit()

    r = await client.get(BUBBLE, headers=rh)
    assert r.status_code == 200
    assert len(r.json()["data"]["items"]) == 1


async def test_settings_are_clamped_and_the_server_echo_is_authoritative(
    client, security_officer,
):
    officer, oh = security_officer
    r = await client.put(
        SETTINGS, json={"daily_new_materials": 999}, headers=oh,
    )
    # Out of the declared range → rejected at the edge rather than
    # silently clamped, so the admin knows the value did not take.
    assert r.status_code == 422

    r = await client.put(SETTINGS, json={"daily_new_materials": 5}, headers=oh)
    assert r.status_code == 200
    assert r.json()["data"]["daily_new_materials"] == 5


async def test_employees_cannot_read_or_change_the_caps(
    client, security_officer, reader,
):
    officer, _oh = security_officer
    _person, rh = reader
    assert (await client.get(SETTINGS, headers=rh)).status_code == 403
    assert (await client.put(
        SETTINGS, json={"enabled": False}, headers=rh,
    )).status_code == 403


async def test_management_sees_the_caps_read_only(
    client, db_session, security_officer,
):
    """They are entitled to know how hard the workforce is being nudged;
    they are not entitled to change it."""
    boss = await make_person(db_session, "boss3@test.com", role="management")
    await db_session.commit()
    bh = headers_for(boss, ["management"])

    r = await client.get(SETTINGS, headers=bh)
    assert r.status_code == 200
    assert r.json()["data"]["can_manage"] is False
    assert (await client.put(
        SETTINGS, json={"enabled": False}, headers=bh,
    )).status_code == 403