"""Completion contracts use fake browser evidence in isolated SQLite/PG stores."""
from __future__ import annotations

import copy

import pytest

from apps.server.db import digest
from apps.server.learning_checks import output_assertion_checks
from apps.server.tests.test_learning import client, initialize, working_candidate
from apps.server.tests.test_learning_extensions import EXTRACT, extract_output
from apps.server.tests.test_postgres import pg_client


@pytest.fixture(params=["client", "pg_client"])
def completion_client(request):
    return request.getfixturevalue(request.param)


def test_scroll_completion_contract_freezes_and_replays_without_rewriting_legacy(completion_client):
    client = completion_client
    initialize(client)
    task, old_candidate = working_candidate(client)
    prefix = "/api/learning/tasks/" + task["id"]
    worker = client.worker_reference
    original = worker.call

    async def completed(method, params):
        result = await original(method, params)
        if method == "action" and params["kind"] == "extract":
            result["result"] = {**extract_output(), "truncated": True, "skipped_incomplete_count": 1}
        elif method == "verify":
            result["outputs"]["structured"] = {**extract_output(), "truncated": True, "skipped_incomplete_count": 1}
        return result
    worker.call = completed
    scroll = {"kind": "scroll", "selector": "body", "pixels": 320, "max_scrolls": 4, "until_selector": '[data-testid="comment"]'}
    extract = {**EXTRACT, "complete_only": True}
    for action in (scroll, extract):
        response = client.post(prefix + "/browser/action", json=action)
        assert response.status_code == 200, response.text
    candidate = client.post(prefix + "/candidates", json={}).json()
    assert candidate["manifest"]["steps"][-2]["until_selector"] == scroll["until_selector"]
    assert candidate["manifest"]["steps"][-1]["complete_only"] is True
    assert client.post("/api/learning/candidates/" + candidate["id"] + "/check").json()["status"] == "passed"
    verify = client.post("/api/learning/candidates/" + candidate["id"] + "/verify", json={"mode": "exploration", "expected_outputs": {"structured": "建材评论"}})
    assert verify.status_code == 200 and verify.json()["status"] == "passed", verify.text
    assert verify.json()["actual_outputs"]["structured"]["skipped_incomplete_count"] == 1
    wire = worker.calls[-1][1]
    assert wire["steps"][-2]["until_selector"] == scroll["until_selector"] and wire["steps"][-1]["complete_only"] is True
    # Existing frozen versions neither gain fields nor change their digest.
    stored = client.get("/api/learning/candidates/" + old_candidate["id"]).json()
    assert stored["manifest"] == old_candidate["manifest"] and stored["digest"] == old_candidate["digest"]
    assert client.post("/api/learning/candidates/" + old_candidate["id"] + "/check").json()["status"] == "passed"
    for bad in (
        {**scroll, "until_selector": ""}, {**scroll, "until_selector": " "},
        {**scroll, "until_selector": "body .body"}, {**scroll, "until_selector": "html"},
        {**scroll, "until_selector": ":root .body"}, {**scroll, "until_selector": "xpath=//div"},
        {**scroll, "until_selector": "javascript:alert(1)"}, {**scroll, "until_selector": "x" * 1001},
        {**scroll, "until_selector": None},
        {**EXTRACT, "until_selector": ".body"}, {**scroll, "complete_only": True},
        {**extract, "complete_only": 1}, {**extract, "complete_only": "true"},
        {**extract, "complete_only": None},
    ):
        count = len([call for call in worker.calls if call[0] == "action"])
        assert client.post(prefix + "/browser/action", json=bad).status_code == 422
        assert len([call for call in worker.calls if call[0] == "action"]) == count
    service = client.app_reference.state.learning
    malformed = copy.deepcopy(candidate)
    malformed["manifest"]["steps"][-2]["until_selector"] = "body .body"
    malformed["digest"] = digest(malformed["manifest"])
    assert any("契约" in error for error in service.static_errors(malformed))


def test_complete_only_shape_is_explicit_and_never_passes_empty_or_missing_fields():
    contract = {**EXTRACT, "complete_only": True}
    good = {**extract_output(), "skipped_incomplete_count": 2, "truncated": True}
    def checks(value, current=contract):
        return output_assertion_checks({"structured": value}, {"structured": "建材评论"}, {"structured": current})
    assert all(item["passed"] for item in checks(good))
    assert all(item["passed"] for item in checks({**good, "skipped_incomplete_count": 0, "truncated": False}))
    # Legacy extracts continue to use exactly the old five-field shape.
    assert all(item["passed"] for item in checks(extract_output(), EXTRACT))
    invalid = [extract_output(), {**good, "skipped_incomplete_count": -1}, {**good, "skipped_incomplete_count": True},
        {**good, "skipped_incomplete_count": 1.5}, {**good, "truncated": False},
        {**good, "items": [{"body": None, "link": None}]}, {**good, "items": [{"link": None}]},
        {**good, "count": 2}, {**good, "unexpected": "payload"}]
    for value in invalid:
        assert not checks(value)[0]["passed"]
    assert not checks(good, EXTRACT)[0]["passed"]
    empty = {"items": [], "count": 0, "truncated": True, "coverage": "empty", "source_url": good["source_url"], "skipped_incomplete_count": 2}
    result = checks(empty)
    assert result[0]["passed"] and not all(item["passed"] for item in result)
    assert not next(item for item in result if item["name"] == "output_present:structured")["passed"]