import asyncio
from datetime import timedelta
from uuid import UUID

import pytest
from sqlalchemy import func, select
from test_after_sales import scenario  # noqa: F401
from test_conversations import setup

from zhifu.modules.identity.access_models import StoreGrant
from zhifu.modules.identity.auth import now
from zhifu.modules.merchant.models import ChannelApplication
from zhifu.modules.notifications.models import Notification
from zhifu.modules.storefront.models import CatalogCategory, CatalogProduct, IdentityChallenge

pytestmark = pytest.mark.integration


async def prepare_channel(context):
    # 独立测试租户只验证身份应用服务；回调传输由平台 TLS 接收器测试覆盖。
    async with context["app"].state.sessions.begin() as session:
        session.add(ChannelApplication(
            tenant_id=context["tenant"].bytes, store_id=context["store"].bytes,
            name="独立身份组件验证渠道", callback_url="https://127.0.0.1/channel-component",
            enabled=True, version=1, scopes=["tickets:results"], updated_at=now(),
        ))
        for permission in ["identities:verify", "tickets:read"]:
            session.add(StoreGrant(
                tenant_id=context["tenant"].bytes, store_id=context["store"].bytes,
                principal_id=context["accounts"]["connector"].bytes, permission=permission,
            ))


async def challenge(context, actor="customer", key=None):
    response = await context["call"]("POST", "/openapi/v1/me/identity-challenges", actor,
                                     {"store_id": str(context["store"]), "channel": "webhook"}, key)
    assert response.status_code == 200, response.text
    return response.json()


def proof(context, external="verified-channel-account"):
    return {"store_id": str(context["store"]), "channel": "webhook",
            "external_subject": external, "display_name": "经过渠道认证的客户"}


async def test_catalog_order_metadata_and_visibility(scenario):  # noqa: F811
    sessions = scenario["app"].state.sessions
    tenant, store = scenario["tenant"].bytes, scenario["store"].bytes
    async with sessions.begin() as session:
        session.add(CatalogCategory(tenant_id=tenant, store_id=store,
                                    code="keyboard", name="键盘", position=0))
        await session.flush()
        for code, status in [("keyboard-1", "ACTIVE"), ("keyboard-2", "ACTIVE"),
                             ("keyboard-3", "HIDDEN")]:
            session.add(CatalogProduct(
                tenant_id=tenant, store_id=store, id=code, category_code="keyboard",
                name="办公键盘", description="日常文字输入", specifications={"款式": "标准款"},
                delivery="店铺配送", image_url="/products/keyboard-01.jpg",
                image_source="https://www.pexels.com/photo/17479949/", status=status,
            ))
    client = scenario["client"]
    path = f"/openapi/v1/stores/{scenario['store']}"
    first = await client.get(path + "/catalog?limit=1")
    assert first.status_code == 200
    assert first.json()["next_cursor"] == "keyboard-1"
    second = await client.get(path + "/catalog?limit=1&cursor=keyboard-1")
    assert second.json()["next_cursor"] is None
    assert second.json()["items"][0]["id"] == "keyboard-2"
    assert (await client.get(path + "/products/keyboard-3")).status_code == 404
    assert (await client.get(path + "/catalog?category=missing")).json()["items"] == []
    order = await scenario["call"]("GET", f"/openapi/v1/orders/{scenario['order_id']}")
    assert order.json()["product_name"] == "办公键盘"
    assert order.json()["specifications"] == {"款式": "标准款"}
    assert order.json()["fulfillment_status"] == "SIGNED"
    assert order.json()["available_quantity"] == 1
    orders_path = f"/openapi/v1/orders?store_id={scenario['store']}"
    listed = await scenario["call"]("GET", orders_path)
    assert listed.status_code == 200, listed.text
    assert listed.json()["items"][0]["order_id"] == str(scenario["order_id"])
    assert listed.json()["items"][0]["available_quantity"] == 1
    foreign = await scenario["call"]("GET", f"/openapi/v1/orders/{scenario['order_id']}", "foreign")
    assert foreign.status_code == 404
    # 来源更新到零时，列表与详情都应显示最新可售后数量。
    synchronized = await scenario["call"](
        "PUT", f"/internal/v1/orders/{scenario['order_id']}", "connector",
        {**scenario["source"], "source_version": 2, "available_quantity": 0},
    )
    assert synchronized.status_code == 200, synchronized.text
    listed = await scenario["call"]("GET", orders_path)
    order = await scenario["call"]("GET", f"/openapi/v1/orders/{scenario['order_id']}")
    assert listed.json()["items"][0]["available_quantity"] == 0
    assert order.json()["available_quantity"] == 0


async def test_handoff_transaction_retry_and_takeover(scenario):  # noqa: F811
    path = await setup(scenario)
    call = scenario["call"]
    body = {"expected_control_version": 1}
    requests = await asyncio.gather(*[
        call("POST", path + "/human-assistance", payload=body, key="handoff-same")
        for _ in range(3)
    ])
    assert all(response.status_code == 200 for response in requests)
    assert requests[0].json() == requests[1].json() == requests[2].json()
    assert requests[0].json()["requested"]
    assert (await call("GET", path + "/human-assistance")).json()["requested"]
    assert (await call("GET", path + "/human-assistance", "other")).status_code == 404
    conflict = await call("POST", path + "/human-assistance",
                          payload={"expected_control_version": 9}, key="handoff-same")
    assert conflict.json()["code"] == "IDEMPOTENCY_CONFLICT"
    async with scenario["app"].state.sessions() as session:
        count = await session.scalar(select(func.count()).select_from(Notification).where(
            Notification.tenant_id == scenario["tenant"].bytes,
        ))
        assert count == 2
    assert (await call("POST", path + "/takeover", "agent", body)).status_code == 200
    assert not (await call("GET", path + "/human-assistance")).json()["requested"]
    assert (await call("POST", path + "/release", "agent",
                       {"expected_control_version": 2})).status_code == 200
    assert not (await call("GET", path + "/human-assistance")).json()["requested"]
    fresh = await call("POST", path + "/human-assistance",
                       payload={"expected_control_version": 3})
    assert fresh.status_code == 200 and fresh.json()["requested"]


async def test_identity_proof_confirmation_isolation_and_conflict(scenario):  # noqa: F811
    await prepare_channel(scenario)
    call = scenario["call"]
    first = await challenge(scenario, key="challenge-same")
    assert await challenge(scenario, key="challenge-same") == first
    identifier = first["challenge_id"]
    path = f"/openapi/v1/me/identity-challenges/{identifier}"
    verify = f"/openapi/v1/channel/identity-challenges/{identifier}/verify"
    assert (await call("POST", path + "/confirm")).json()["code"] == "IDENTITY_NOT_VERIFIED"
    for actor in ["other", "foreign"]:
        assert (await call("GET", path, actor)).status_code == 404
        assert (await call("POST", path + "/confirm", actor)).status_code == 404
    assert (await call("POST", verify, payload=proof(scenario))).status_code == 403
    response = await call("POST", verify, "connector", proof(scenario))
    assert response.status_code == 200 and response.json()["status"] == "VERIFIED"
    assert (await call("POST", verify, "connector", proof(scenario))).json() == response.json()
    conflict = await call("POST", verify, "connector", proof(scenario, "other-account"))
    assert conflict.status_code == 409 and conflict.json()["code"] == "IDENTITY_CONFLICT"
    confirmed = await asyncio.gather(*[call("POST", path + "/confirm") for _ in range(3)])
    assert all(item.status_code == 200 and item.json()["status"] == "LINKED"
               for item in confirmed)
    identities = await call(
        "GET", f"/openapi/v1/me/channel-identities?store_id={scenario['store']}"
    )
    assert len(identities.json()["items"]) == 1
    other = await challenge(scenario, "other")
    other_id = other["challenge_id"]
    assert (await call("POST", f"/openapi/v1/channel/identity-challenges/{other_id}/verify",
                       "connector", proof(scenario))).status_code == 200
    collision = await call(
        "POST", f"/openapi/v1/me/identity-challenges/{other_id}/confirm", "other"
    )
    assert collision.status_code == 409 and collision.json()["code"] == "IDENTITY_CONFLICT"
    tickets = f"/openapi/v1/channel/identities/tickets?store_id={scenario['store']}"
    assert (await call("GET", tickets + "&external_subject=verified-channel-account",
                       "connector")).status_code == 200
    missing = await call("GET", tickets + "&external_subject=unknown", "connector")
    assert missing.status_code == 404


async def test_identity_expiration_and_channel_disabled(scenario):  # noqa: F811
    await prepare_channel(scenario)
    call = scenario["call"]
    pending = await challenge(scenario)
    identifier = pending["challenge_id"]
    path = f"/openapi/v1/me/identity-challenges/{identifier}"
    verify = f"/openapi/v1/channel/identity-challenges/{identifier}/verify"
    assert (await call("POST", verify, "connector", proof(scenario))).status_code == 200
    async with scenario["app"].state.sessions.begin() as session:
        row = await session.get(IdentityChallenge, UUID(identifier).bytes)
        row.expires_at = now() - timedelta(seconds=1)
    assert (await call("POST", path + "/confirm")).json()["code"] == "CHALLENGE_EXPIRED"
    assert (await call("POST", verify, "connector",
                       proof(scenario))).json()["code"] == "CHALLENGE_EXPIRED"
    async with scenario["app"].state.sessions.begin() as session:
        channel = await session.get(ChannelApplication,
                                    (scenario["tenant"].bytes, scenario["store"].bytes))
        channel.enabled = False
    body = {"store_id": str(scenario["store"]), "channel": "webhook"}
    disabled = await call("POST", "/openapi/v1/me/identity-challenges", payload=body)
    assert disabled.status_code == 409 and disabled.json()["code"] == "CHANNEL_UNAVAILABLE"