"""Check UI stream ownership without changing or mocking comparison semantics."""

from copy import deepcopy

import pytest

from demo.presentation import json_detail, mismatch_summary, status
from demo.session import UISession, begin_ui, cancel_session, edit_inputs, reset_session, stream_ui

pytestmark = pytest.mark.demo


def streamed_handler(closed):
    def handler(*values):
        try:
            yield status("Running 1/5"), [["0"]], None, None, "original", "candidate"
            yield status("Completed · AGREE", "green"), [["0"], ["1"]], None, "report.json", "original", "candidate"
        finally:
            closed.append(True)
    return handler


def test_edit_during_stream_preserves_snapshot_staleness_and_download():
    session, closed = UISession(), []
    iterator = stream_ui(streamed_handler(closed), ("source", "settings"), session)
    assert "Building" in next(iterator)[0]
    running = next(iterator)
    assert "Running" in running[0]
    assert "Out of date" in edit_inputs(session, "edited", "settings")
    final = next(iterator)
    assert "Completed" in final[0] and "Out of date" in final[0]
    assert final[1:] == ([["0"], ["1"]], None, "report.json", "original", "candidate")
    assert session.snapshot == ("source", "settings")
    assert list(iterator) == []
    assert not session.busy and closed == [True]
    # Returning to the displayed inputs makes the existing result current again.
    assert "Out of date" not in edit_inputs(session, "source", "settings")


def test_cancel_rejects_late_updates_and_closes_original_stream():
    session, closed = UISession(), []
    iterator = stream_ui(streamed_handler(closed), ("source",), session)
    next(iterator)
    next(iterator)
    edit_inputs(session, "edited")
    cancelled = cancel_session(session)
    assert "Cancelled" in cancelled and "Out of date" in cancelled
    assert list(iterator) == []
    assert closed == [True] and not session.busy
    assert session.rendered() == cancelled


@pytest.mark.parametrize("updates", [1, 2], ids=["building", "running"])
def test_closing_ui_stream_releases_session_and_started_handler(updates):
    session, closed = UISession(), []
    iterator = stream_ui(streamed_handler(closed), ("source",), session)
    for _ in range(updates):
        next(iterator)
    assert session.busy
    iterator.close()
    assert not session.busy
    assert closed == ([True] if updates == 2 else [])


def test_example_reset_rejects_previous_stream_without_staling_new_results():
    session, closed = UISession(), []
    old = stream_ui(streamed_handler(closed), ("first example",), session)
    next(old)
    next(old)
    reset_session(session)
    assert session.rendered() == "" and session.snapshot is None
    new = stream_ui(streamed_handler([]), ("second example",), session)
    next(new)
    assert list(old) == [] and closed == [True]
    assert session.busy
    updates = list(new)
    assert "Completed" in updates[-1][0] and "Out of date" not in updates[-1][0]
    assert not session.busy


def test_preview_edit_and_independent_sessions():
    first = UISession()
    second = deepcopy(first)
    assert first.lock is not second.lock
    def preview(*args):
        yield status("Completed · Assembly preview"), [], None, None, "a", "b"
    list(stream_ui(preview, ("source",), first))
    assert "Out of date" in edit_inputs(first, "edited")
    assert second.snapshot is None and second.rendered() == ""


def test_queued_work_uses_click_snapshot_and_rejects_cancelled_ticket():
    session = UISession()
    ticket = begin_ui(session, "clicked source")[-1]
    edit_inputs(session, "edited before queue started")
    received = []

    def handler(*values):
        received.append(values)
        yield status("Completed"), [], None, None, "a", "b"

    final = list(stream_ui(handler, ("edited before queue started",), session, ticket))[-1]
    assert received == [("clicked source",)]
    assert "Out of date" in final[0]
    ticket = begin_ui(session, "another source")[-1]
    cancel_session(session)
    assert list(stream_ui(handler, ("another source",), session, ticket)) == []


def test_first_mismatch_summary_preserves_values_and_escapes_text():
    detail = {"seed": 42, "reason": "<bad>", "difference": {
        "location": "rax<script>", "original": 0x102, "candidate": 0x203,
        "bit_ranges": [{"first": 0, "last": 0}, {"first": 8, "last": 9}]}}
    rendered = mismatch_summary(detail)
    for expected in ("42", "0x102", "0x203", "0, 8–9", "rax&lt;script&gt;", "&lt;bad&gt;"):
        assert expected in rendered
    assert "<script>" not in rendered
    assert "No mismatch" in mismatch_summary(None)


@pytest.mark.parametrize("value", [0x7ff8000000000001, (1 << 127) + 3, 1 << 16000],
                         ids=["nan-payload", "128-bit", "unbounded"])
def test_mismatch_widget_preserves_integer_bits(value):
    import gradio as gr
    from verifier.serialization import integer_text

    detail = {"seed": 0, "difference": {"location": "return", "width": value.bit_length(),
              "original": value, "candidate": value ^ 1,
              "bit_ranges": [{"first": 0, "last": 0}]}}
    displayed = gr.JSON().postprocess(json_detail(detail)).root
    # Strings retain their digits through the browser, including NaN payloads.
    assert displayed["difference"]["original"] == integer_text(value)
    assert displayed["difference"]["candidate"] == integer_text(value ^ 1)
    assert hex(value) in mismatch_summary(displayed)