"""Regression for #1653 — Feishu v1 body-token webhook replay protection.

The v1 (``approval_instance`` v1.0) callback path has no signature header
and no nonce; it authenticates via the static ``event_verification_token``
in the JSON body plus a mandatory ``X-Lark-Request-Timestamp`` header.

The Round-49 SETNX replay cache on this path hashed an undefined
``content`` variable (``content`` is only bound in the v2 branch), so on
the v1 path ``verify_webhook`` raised ``NameError`` on EVERY call. The
broad ``except`` swallowed it and fell through to ``ok=True`` — the entire
v1 replay cache was dead code and fail-open: a captured approval callback
could be replayed indefinitely within the ±300s window.

Post-fix the driver hashes the raw ``body`` param, so a second delivery of
the SAME body + timestamp is rejected as a replay.

Uses the real Redis in the isolated test env (``get_redis``). A unique
body per test run keeps the SETNX key from colliding across runs.
"""
from __future__ import annotations

import json
import time
import uuid

import pytest

from app.core.redis_client import get_redis
from app.integrations.drivers.feishu_chat_approval import (
    FeishuChatApprovalDriver,
)

_TOKEN = "verif-token-123"
_CONFIG = {
    "app_id": "cli_test",
    "app_secret": "secret",
    "event_verification_token": _TOKEN,
    # No event_encrypt_key → forces the v1 body-token path.
}


def _driver() -> FeishuChatApprovalDriver:
    return FeishuChatApprovalDriver(integration_id="feishu-test", config=_CONFIG)


def _v1_body() -> bytes:
    """A v1 approval-instance callback body with a unique uuid so the
    body-hash SETNX key is fresh for this test run (avoids collisions
    with a prior run's leftover key inside the 600s TTL)."""
    return json.dumps({
        "token": _TOKEN,
        "type": "event_callback",
        "event": {
            "type": "approval_instance",
            "instance_code": f"inst-{uuid.uuid4()}",
        },
    }).encode()


@pytest.fixture(autouse=True)
async def _flush_replay_keys():
    """Best-effort cleanup of this test's replay keys before + after so a
    prior run's key can't make the 'first call' spuriously look replayed."""
    r = get_redis()
    yield
    try:
        async for key in r.scan_iter("feishu:webhook_replay_v1:*"):
            await r.delete(key)
    except Exception:  # noqa: BLE001 — cleanup is best-effort
        pass


@pytest.mark.asyncio
async def test_v1_body_token_replay_rejected_on_second_delivery() -> None:
    body = _v1_body()
    ts = str(int(time.time()))
    headers = {"x-lark-request-timestamp": ts}

    drv = _driver()
    first = await drv.verify_webhook(body=body, headers=headers)
    assert first.ok is True, first.error_message
    assert first.event is not None

    # Identical body + timestamp = replay. Pre-fix this was ok=True
    # (NameError swallowed, fail-open); post-fix the body-hash SETNX
    # cache rejects it.
    second = await drv.verify_webhook(body=body, headers=headers)
    assert second.ok is False
    assert "repla" in (second.error_message or "").lower()


@pytest.mark.asyncio
async def test_v1_distinct_bodies_both_accepted() -> None:
    """Two genuinely-different callbacks (distinct bodies) at the same
    timestamp are BOTH accepted — the dedup keys on body-hash, so it
    must not false-positive on unrelated events."""
    ts = str(int(time.time()))
    headers = {"x-lark-request-timestamp": ts}
    drv = _driver()

    r1 = await drv.verify_webhook(body=_v1_body(), headers=headers)
    r2 = await drv.verify_webhook(body=_v1_body(), headers=headers)
    assert r1.ok is True
    assert r2.ok is True