"""Bounded request-denial reasons and new immutable failure readback.

No provider or platform is contacted: API checks use a fake worker and either
temporary SQLite or the opt-in dedicated PostgreSQL test DB's random schema.
"""
from __future__ import annotations

import copy
import json

import pytest

from apps.server.tests.test_discovery import item, search_candidate
from apps.server.tests.test_learning import client, initialize, working_candidate
from apps.server.tests.test_postgres import pg_client
from apps.server.tests.test_reading import NOTE_ID, TITLE, detail_output, learned_detail
from apps.server.worker import (
    ACTIVE_REQUEST_POLICIES, REQUEST_POLICY_REASONS, WorkerError,
    safe_diagnostics, safe_failure_diagnostics, safe_request_observations, safe_request_summary,
)

VERSION = "xhs-readonly-ui-experimental-v1"
REQUEST = {"method": "POST", "origin": "https://apm-fe.xiaohongshu.com", "pathname": "/api/data?query=PRIVATE_FIXTURE#PRIVATE_FIXTURE",
    "resource_type": "xhr", "policy_reason": "resource_type_not_allowed", "active_policy_version": VERSION,
    "expected_resource_types": ["fetch", "fetch"], "headers": {"Cookie": "PRIVATE_FIXTURE"}, "body": "PRIVATE_FIXTURE", "query": "PRIVATE_FIXTURE"}
DIAGNOSTICS = {"phase": "execute", "blocked_requests": [REQUEST], "blocked_request_count": 1, "stack": "PRIVATE_FIXTURE"}


def failure(position=2, kind="click", code="WRITE_REQUEST_BLOCKED"):
    return {"step_position": position, "kind": kind, "code": code, "diagnostics": copy.deepcopy(DIAGNOSTICS)}


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


def test_request_denials_keep_only_closed_reasons_current_versions_and_safe_types():
    expected_base = {"method": "POST", "origin": "https://apm-fe.xiaohongshu.com", "pathname": "/api/data", "resource_type": "xhr"}
    assert safe_request_summary(REQUEST) == expected_base
    actual = safe_request_summary(REQUEST, blocked=True)
    assert actual == expected_base | {"policy_reason": "resource_type_not_allowed", "active_policy_version": VERSION, "expected_resource_types": ["fetch"]}
    for reason in REQUEST_POLICY_REASONS:
        value = safe_request_summary(REQUEST | {"policy_reason": reason}, blocked=True)
        assert value["policy_reason"] == reason
        assert ("expected_resource_types" in value) == (reason == "resource_type_not_allowed")
    assert len(REQUEST_POLICY_REASONS) == 15 and len(ACTIVE_REQUEST_POLICIES) == 4
    for version in ACTIVE_REQUEST_POLICIES:
        assert safe_request_summary(REQUEST | {"active_policy_version": version}, blocked=True)["active_policy_version"] == version
    for version in ("xhs-search-experimental-v1", "xhs-search-experimental-v2", "xhs-note-open-experimental-v1", "PRIVATE_FIXTURE", [], None):
        assert "active_policy_version" not in safe_request_summary(REQUEST | {"active_policy_version": version}, blocked=True)
    for reason in ("PRIVATE_FIXTURE", [], None, 2):
        value = safe_request_summary(REQUEST | {"policy_reason": reason}, blocked=True)
        assert "policy_reason" not in value and "expected_resource_types" not in value
    for types in (None, "fetch", {}, [], [[], {}, 1, True, "PRIVATE_FIXTURE"]):
        assert "expected_resource_types" not in safe_request_summary(REQUEST | {"expected_resource_types": types}, blocked=True)
    value = safe_request_summary(REQUEST | {"expected_resource_types": ["fetch", "fetch", "xhr", "image", "other", "font", "PRIVATE_FIXTURE", {}]}, blocked=True)
    assert value["expected_resource_types"] == ["fetch", "xhr", "image", "other"]
    assert "PRIVATE_FIXTURE" not in json.dumps(value)


def test_observation_denials_require_actual_boolean_block_and_keep_allowed_rule_contract():
    base = REQUEST | {"automating": True, "blocked": True, "sequence": 2, "recorded_at": "2026-10-25T09:00:00.000Z"}
    blocked = safe_request_observations([base])[0]
    assert blocked["policy_reason"] == "resource_type_not_allowed" and blocked["active_policy_version"] == VERSION
    assert blocked["expected_resource_types"] == ["fetch"]
    for block in (False, "true", 1, None):
        value = safe_request_observations([base | {"blocked": block}])[0]
        assert not {"policy_reason", "active_policy_version", "expected_resource_types"} & value.keys()
    allowed = safe_request_observations([base | {"blocked": False, "resource_type": "fetch", "policy_version": VERSION, "policy_rule": "apm_data"}])[0]
    assert allowed["policy_version"] == VERSION and allowed["policy_rule"] == "apm_data"
    assert not {"policy_reason", "active_policy_version", "expected_resource_types"} & allowed.keys()
    # Adding denial diagnostics cannot add an allowed label to a blocked request.
    rejected = safe_request_observations([base | {"policy_version": VERSION, "policy_rule": "apm_data"}])[0]
    assert "policy_version" not in rejected and "policy_rule" not in rejected
    assert "PRIVATE_FIXTURE" not in json.dumps([blocked, allowed, rejected])


def test_diagnostics_bounds_and_repeated_sanitization_never_restore_raw_payload():
    value = safe_diagnostics(DIAGNOSTICS | {"blocked_requests": [REQUEST] * 30, "blocked_request_count": 2_000_000})
    assert len(value["blocked_requests"]) == 12 and value["blocked_request_count"] == 1_000_000
    assert safe_diagnostics(value) == value
    assert value["phase"] == "execute" and "PRIVATE_FIXTURE" not in json.dumps(value)
    assert safe_diagnostics({"phase": [], "blocked_requests": [None, {"method": []}]}) == {"phase": "unknown", "blocked_requests": [], "blocked_request_count": 0}
    e = WorkerError("Not persisted", "WRITE_REQUEST_BLOCKED", diagnostics=DIAGNOSTICS)
    assert e.diagnostics == safe_diagnostics(DIAGNOSTICS)


def test_failure_diagnostics_needs_real_failed_code_and_exact_step_not_new_outcome():
    steps = [{"kind": "fill"}, {"kind": "click"}, {"kind": "read"}]
    codes = ["WRITE_REQUEST_BLOCKED"]
    cleaned = safe_failure_diagnostics(failure(), steps, codes, passed=False)
    assert cleaned == failure() | {"diagnostics": safe_diagnostics(DIAGNOSTICS)}
    assert "PRIVATE_FIXTURE" not in json.dumps(cleaned)
    for passed in (True, None, "false", 0):
        assert safe_failure_diagnostics(failure(), steps, codes, passed=passed) is None
    for changes in ({"step_position": True}, {"step_position": 0}, {"step_position": 4}, {"step_position": "2"},
                    {"kind": "read"}, {"kind": []}, {"code": "WAIT_TIMEOUT"}, {"code": []},
                    {"diagnostics": None}, {"extra": "PRIVATE_FIXTURE"}):
        assert safe_failure_diagnostics(failure() | changes, steps, codes, passed=False) is None
    assert safe_failure_diagnostics(failure(), steps, [], passed=False) is None
    assert safe_failure_diagnostics(None, steps, codes, passed=False) is None


def test_failed_recording_keeps_denial_reasons_through_both_error_boundaries(diagnostics_client):
    client = diagnostics_client
    initialize(client)
    task, old = working_candidate(client)
    worker, service = client.worker_reference, client.app_reference.state.learning
    prefix = "/api/learning/tasks/" + task["id"]
    original = worker.call
    steps = client.get(prefix).json()["steps"]
    async def failed(method, params):
        if method == "action":
            after = worker.capture(task["id"])
            after["request_observations"] = [REQUEST | {"automating": True, "blocked": True}]
            raise WorkerError("not returned", "WRITE_REQUEST_BLOCKED", before=worker.capture(task["id"]), after=after, diagnostics=DIAGNOSTICS)
        return await original(method, params)
    worker.call = failed
    result = client.post(prefix + "/browser/action", json={"kind": "click", "selector": '[data-testid="search-button"]'})
    assert result.status_code == 409, result.text
    assert result.json()["detail"]["action_failure"]["diagnostics"] == safe_diagnostics(DIAGNOSTICS)
    assert client.get(prefix).json()["steps"] == steps
    observed = client.get(prefix).json()["observations"][-1]
    assert observed["request_observations"][0]["policy_reason"] == "resource_type_not_allowed"
    assert "PRIVATE_FIXTURE" not in result.text and "PRIVATE_FIXTURE" not in json.dumps(observed)
    current = client.get("/api/learning/candidates/" + old["id"]).json()
    assert current["manifest"] == old["manifest"] and current["digest"] == old["digest"]


def test_verification_appends_only_valid_failure_diagnostics_keeps_legacy_and_success(diagnostics_client):
    client = diagnostics_client
    initialize(client)
    task, candidate = working_candidate(client)
    worker, store = client.worker_reference, client.app_reference.state.store
    endpoint = "/api/learning/candidates/" + candidate["id"]
    assert client.post(endpoint + "/check").json()["status"] == "passed"
    values = {"expected_outputs": {"results": "建材"}}
    legacy = client.post(endpoint + "/verify", json=values).json()
    assert legacy["status"] == "passed" and "failure_diagnostics" not in legacy
    legacy_raw = store.one("SELECT data_json FROM validations WHERE id=$1", (legacy["id"],))
    frozen = store.one("SELECT manifest_json,digest FROM candidates WHERE id=$1", (candidate["id"],))
    mode = {"value": "valid"}
    original = worker.call
    async def diagnosed(method, params):
        value = await original(method, params)
        if method == "verify":
            value.update(passed=False, errors=["本机合成的请求阻断"], error_codes=["WRITE_REQUEST_BLOCKED"], failure_diagnostics=failure())
            value["evidence"][0]["request_observations"] = [REQUEST | {"automating": True, "blocked": True}]
            if mode["value"] == "invalid":
                value["failure_diagnostics"]["kind"] = "read"
            elif mode["value"] == "missing":
                value.pop("failure_diagnostics")
            elif mode["value"] == "passed":
                value.update(passed=True, errors=[], error_codes=[])
        return value
    worker.call = diagnosed
    for current in ("valid", "invalid", "missing", "passed"):
        mode["value"] = current
        response = client.post(endpoint + "/verify", json=values)
        assert response.status_code == 200, response.text
        record = response.json()
        assert record["status"] == ("passed" if current == "passed" else "failed")
        assert ("failure_diagnostics" in record) == (current == "valid")
        if current == "valid":
            assert record["failure_diagnostics"] == failure() | {"diagnostics": safe_diagnostics(DIAGNOSTICS)}
        assert "PRIVATE_FIXTURE" not in response.text
        assert any(value["id"] == record["id"] and value == record for value in client.get(endpoint).json()["validations"])
    assert store.one("SELECT data_json FROM validations WHERE id=$1", (legacy["id"],)) == legacy_raw
    assert store.one("SELECT manifest_json,digest FROM candidates WHERE id=$1", (candidate["id"],)) == frozen
    assert next(value for value in client.get(endpoint).json()["validations"] if value["id"] == legacy["id"]) == legacy


def test_reading_copies_new_failure_top_level_and_preserves_old_records(diagnostics_client):
    client = diagnostics_client
    initialize(client)
    _, search = search_candidate(client, [item(TITLE)])
    run = client.post("/api/discovery/runs", json={"search_candidate_id": search["id"], "keyword": "装修"}).json()
    task, candidate, _ = learned_detail(client)
    worker, store = client.worker_reference, client.app_reference.state.store
    endpoint = "/api/discovery/runs/" + run["id"] + "/readings"
    payload = {"detail_candidate_id": candidate["id"], "note_id": NOTE_ID}
    legacy = client.post(endpoint, json=payload).json()
    assert legacy["status"] == "completed" and "failure_diagnostics" not in legacy
    originals = {"reading": store.one("SELECT data_json FROM reading_runs WHERE id=$1", (legacy["id"],)),
        "validation": store.one("SELECT data_json FROM validations WHERE id=$1", (legacy["validation_id"],)),
        "candidate": store.one("SELECT digest,manifest_json FROM candidates WHERE id=$1", (candidate["id"],)),
        "discovery": store.one("SELECT data_json FROM discovery_runs WHERE id=$1", (run["id"],))}
    mode = {"partial": False}
    original = worker.call
    async def diagnosed(method, params):
        value = await original(method, params)
        if method == "verify":
            value.update(passed=False, errors=["本机合成的请求阻断"], error_codes=["WRITE_REQUEST_BLOCKED"])
            if mode["partial"]:
                value["outputs"] = {"note_detail": detail_output()}
                value["failure_diagnostics"] = failure(7, "extract")
            else:
                value["outputs"] = {}
                value["note_binding_result"] = {"note_id": NOTE_ID, "href_checked": False, "url_checked": False, "source_url": None, "title": None}
                value["failure_diagnostics"] = failure(3, "wait")
        return value
    worker.call = diagnosed
    for partial in (False, True):
        mode["partial"] = partial
        response = client.post(endpoint, json=payload)
        assert response.status_code == 201, response.text
        reading = response.json()
        assert reading["status"] == ("partial" if partial else "failed") and reading["comments"] is None
        assert reading["failure_diagnostics"]["kind"] == ("extract" if partial else "wait")
        assert "PRIVATE_FIXTURE" not in response.text
        validation = next(value for value in client.get("/api/learning/candidates/" + candidate["id"]).json()["validations"] if value["id"] == reading["validation_id"])
        assert validation["status"] == "failed" and validation["failure_diagnostics"] == reading["failure_diagnostics"]
        assert client.get("/api/discovery/readings/" + reading["id"]).json() == reading
    assert store.one("SELECT data_json FROM reading_runs WHERE id=$1", (legacy["id"],)) == originals["reading"]
    assert store.one("SELECT data_json FROM validations WHERE id=$1", (legacy["validation_id"],)) == originals["validation"]
    assert store.one("SELECT digest,manifest_json FROM candidates WHERE id=$1", (candidate["id"],)) == originals["candidate"]
    assert store.one("SELECT data_json FROM discovery_runs WHERE id=$1", (run["id"],)) == originals["discovery"]
    assert client.get("/api/discovery/readings/" + legacy["id"]).json() == legacy