from __future__ import annotations

import json
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from server.agent_core import harness
from server.auth.context import IdentityContext, bind_identity, reset_identity


class _ProjectStore:
    def active_world_key(self) -> str:
        return "shared-approval-project"


@pytest.fixture(autouse=True)
def _isolated_policy_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    import server.state.projects as projects_module

    for env_name in (
        "APS_APPROVAL_ROLE_POLICIES",
        "APS_APPROVAL_INITIATOR_ROLES",
        "APS_APPROVAL_ROLES",
        "APS_APPROVAL_VIEW_ROLES",
    ):
        monkeypatch.delenv(env_name, raising=False)
    monkeypatch.setattr(projects_module, "get_project_store", lambda: _ProjectStore())
    harness.configure_approval_store(str(tmp_path / "approvals.json"))
    yield
    harness._approval_store.clear()


def _identity(user_id: int, role: str) -> IdentityContext:
    return IdentityContext(
        user_id,
        f"{role}-{user_id}",
        f"{role} {user_id}",
        "platform",
        roles=(role,),
    )


def _as(identity: IdentityContext, callback):
    token = bind_identity(identity)
    try:
        return callback()
    finally:
        reset_identity(token)


def _stage(action: str) -> str:
    block = harness.stage_confirmation(
        "role-policy-test",
        action,
        {"versionId": 1},
        title="Role policy test",
        summary_lines=["test"],
    )
    return str(block.props["confirmId"])


def _set_policy(monkeypatch: pytest.MonkeyPatch, policy: dict) -> None:
    monkeypatch.setenv(
        "APS_APPROVAL_ROLE_POLICIES",
        json.dumps(policy, ensure_ascii=True, separators=(",", ":")),
    )


def _policy_entry(action: str) -> dict:
    return next(item for item in harness.list_policy() if item["action"] == action)


def test_exact_action_overrides_power_per_capability(monkeypatch: pytest.MonkeyPatch) -> None:
    _set_policy(monkeypatch, {
        "powers": {
            "P2": {
                "initiate": ["p2-initiator"],
                "approve": ["p2-approver"],
                "history": ["p2-auditor"],
            },
        },
        "actions": {
            "schedule.publish": {
                "approve": ["release-approver"],
            },
        },
    })

    confirm_id = _as(_identity(1, "p2-initiator"), lambda: _stage("schedule.publish"))
    assert _as(
        _identity(2, "p2-approver"),
        lambda: harness.take_confirmation(confirm_id, approve=True),
    ) is None
    approved = _as(
        _identity(3, "release-approver"),
        lambda: harness.take_confirmation(confirm_id, approve=True),
    )
    assert approved and approved["approvalStep"] == 1
    assert _as(_identity(4, "p2-auditor"), harness.list_approval_history)[-1]["status"] == "APPROVED"

    projected = _policy_entry("schedule.publish")["rolePolicy"]
    assert projected["valid"] is True
    assert projected["initiate"] == {
        "roles": ["p2-initiator"], "source": "power", "scope": "P2",
    }
    assert projected["approve"] == {
        "roles": ["release-approver"], "source": "action", "scope": "schedule.publish",
    }
    assert projected["history"] == {
        "roles": ["p2-auditor"], "source": "power", "scope": "P2",
    }


def test_missing_scoped_capabilities_use_existing_global_env_fallback(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("APS_APPROVAL_INITIATOR_ROLES", "global-initiator")
    monkeypatch.setenv("APS_APPROVAL_ROLES", "global-approver")
    monkeypatch.setenv("APS_APPROVAL_VIEW_ROLES", "global-auditor")
    _set_policy(monkeypatch, {
        "actions": {
            "schedule.publish": {"approve": ["release-approver"]},
        },
    })

    confirm_id = _as(_identity(10, "global-initiator"), lambda: _stage("schedule.publish"))
    approved = _as(
        _identity(11, "release-approver"),
        lambda: harness.take_confirmation(confirm_id, approve=True),
    )
    assert approved and approved["needsSecondConfirm"] is False
    assert _as(_identity(12, "global-auditor"), harness.list_approval_history)

    projected = _policy_entry("schedule.publish")["rolePolicy"]
    assert projected["initiate"]["source"] == "global"
    assert projected["initiate"]["scope"] == "APS_APPROVAL_INITIATOR_ROLES"
    assert projected["history"]["roles"] == ["global-auditor"]


@pytest.mark.parametrize(
    "raw_policy",
    [
        "{not-json",
        '{"unknown":{}}',
        '{"actions":{"missing.action":{"approve":["admin"]}}}',
        '{"powers":{"P3":{"approve":"admin"}}}',
        '{"powers":{"P3":{"approve":["admin"],"approve":["auditor"]}}}',
    ],
)
def test_malformed_policy_fails_closed_and_projects_invalid_state(
    raw_policy: str,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("APS_APPROVAL_ROLE_POLICIES", raw_policy)

    with pytest.raises(PermissionError, match="角色权限"):
        _as(_identity(20, "system"), lambda: _stage("mes.dispatch"))
    assert harness._approval_store.pending == {}

    projected = _policy_entry("mes.dispatch")["rolePolicy"]
    assert projected["valid"] is False
    assert projected["systemBypass"] is False
    assert projected["approve"]["roles"] == []
    assert projected["approve"]["source"] == "invalid"


def test_policy_becoming_invalid_blocks_approval_of_existing_pending(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    confirm_id = _as(_identity(25, "planner"), lambda: _stage("schedule.publish"))
    monkeypatch.setenv("APS_APPROVAL_ROLE_POLICIES", '{"powers":{"P2":{"approve":"admin"}}}')

    assert _as(
        _identity(26, "system"),
        lambda: harness.take_confirmation(confirm_id, approve=True),
    ) is None
    assert confirm_id in harness._approval_store.pending
    assert _as(_identity(26, "system"), harness.list_pending) == []


def test_power_policy_keeps_p3_two_person_separation(monkeypatch: pytest.MonkeyPatch) -> None:
    _set_policy(monkeypatch, {
        "powers": {
            "P3": {
                "initiate": ["dispatcher"],
                "approve": ["dispatch-approver"],
                "history": ["dispatch-auditor"],
            },
        },
    })
    confirm_id = _as(_identity(30, "dispatcher"), lambda: _stage("mes.dispatch"))
    first_approver = _identity(31, "dispatch-approver")
    first = _as(first_approver, lambda: harness.take_confirmation(confirm_id, approve=True))
    denied = _as(first_approver, lambda: harness.take_confirmation(confirm_id, approve=True))
    second = _as(
        _identity(32, "dispatch-approver"),
        lambda: harness.take_confirmation(confirm_id, approve=True),
    )

    assert first and first["needsSecondConfirm"] is True
    assert denied and denied["approvalDenied"] is True and denied["separationRequired"] is True
    assert second and second["approvalStep"] == 2 and second["executionGrant"]
    history = _as(_identity(33, "dispatch-auditor"), harness.list_approval_history)
    assert [item["status"] for item in history][-3:] == [
        "PARTIALLY_APPROVED", "SOD_DENIED", "APPROVED",
    ]


def test_policy_api_keeps_existing_shape_and_adds_role_projection(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    import server.gateway.app as gateway_module
    from tests.auth_provider import install_test_auth

    _set_policy(monkeypatch, {
        "actions": {
            "schedule.publish": {
                "approve": ["planner"],
            },
        },
    })
    install_test_auth(monkeypatch, "tenant-role-policy-api")
    client = TestClient(gateway_module.create_app())
    assert client.post("/api/auth/login", json={"username": "planner"}).status_code == 200

    response = client.get("/api/gov/policy")
    assert response.status_code == 200
    payload = response.json()
    assert payload["defaultPower"] == "P3"
    entry = next(item for item in payload["policy"] if item["action"] == "schedule.publish")
    assert {"action", "power", "desc", "confirm"} <= set(entry)
    assert entry["rolePolicy"]["approve"] == {
        "roles": ["planner"], "source": "action", "scope": "schedule.publish",
    }