"""#1714：孤儿消息撤销/清理逻辑测试（mock redis，CI 无真实 redis 时也产生覆盖）。

覆盖 packages/shared/celery_orphan_guard.py：
- _extract_business_ids：三元组 body / 裸 args body / dict args / headers 提取 /
  无 body / 坏 JSON / 坏 base64 / 空 args
- _purge_one_queue：biz id 命中、celery id 命中、未命中保序（重写 rpush）、
  lrange 异常、重写异常、空队列
- purge_stale_messages_from_queues：空 ids 早退、redis 未安装、连接失败、
  正常清理并 close
- revoke_and_purge：revoke 逐消息调用、revoke 异常不阻断、空 id 跳过
- ensure_task_claimable：任务不存在返回空串、终态抛错、pending 放行
"""

from __future__ import annotations

import base64
import json
import sys
import types
from pathlib import Path
from unittest.mock import MagicMock

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

from packages.shared import celery_orphan_guard as guard  # noqa: E402


def _envelope(celery_id: str | None, body_payload) -> bytes:
    """构造 Redis transport 存储的 celery 消息（JSON 信封）。"""
    if body_payload is None:
        body = None
    else:
        body = base64.b64encode(json.dumps(body_payload).encode()).decode()
    envelope = {"body": body, "headers": {"id": celery_id, "task": "worker.generate_video"}}
    return json.dumps(envelope).encode()


# ── _extract_business_ids ───────────────────────────────────────────────


def test_extract_ids_standard_tuple_body():
    raw = _envelope("celery-1", [["biz-task-1"], {}, {"callbacks": None}])
    assert guard._extract_business_ids(raw) == ("celery-1", "biz-task-1")


def test_extract_ids_bare_args_body():
    raw = _envelope("celery-2", ["biz-task-2"])
    assert guard._extract_business_ids(raw) == ("celery-2", "biz-task-2")


def test_extract_ids_dict_body_with_args():
    raw = _envelope("celery-3", {"args": ["biz-task-3"], "kwargs": {}})
    assert guard._extract_business_ids(raw) == ("celery-3", "biz-task-3")


def test_extract_ids_non_dict_headers_returns_celery_id_none():
    raw = json.dumps({"body": base64.b64encode(json.dumps([["biz-4"]]).encode()).decode(), "headers": "x"}).encode()
    celery_id, biz_id = guard._extract_business_ids(raw)
    assert celery_id is None
    assert biz_id == "biz-4"


def test_extract_ids_no_body_returns_celery_id_only():
    raw = json.dumps({"headers": {"id": "celery-5"}}).encode()
    assert guard._extract_business_ids(raw) == ("celery-5", None)


def test_extract_ids_empty_args_returns_no_biz_id():
    raw = _envelope("celery-6", [[], {}, {}])
    assert guard._extract_business_ids(raw) == ("celery-6", None)


def test_extract_ids_args_first_none_returns_no_biz_id():
    raw = _envelope("celery-7", [[None], {}, {}])
    assert guard._extract_business_ids(raw) == ("celery-7", None)


def test_extract_ids_bad_json_returns_none_none():
    assert guard._extract_business_ids(b"not-json{") == (None, None)


def test_extract_ids_bad_base64_returns_none_none():
    raw = json.dumps({"body": "!!!not-base64!!!", "headers": {"id": "c"}}).encode()
    assert guard._extract_business_ids(raw) == (None, None)


def test_extract_ids_int_arg_coerced_to_str():
    raw = _envelope("celery-9", [[12345], {}, {}])
    celery_id, biz_id = guard._extract_business_ids(raw)
    assert celery_id == "celery-9"
    assert biz_id == "12345"


# ── _purge_one_queue ────────────────────────────────────────────────────


def _queue_with_messages(*payloads: bytes):
    """返回 list-backed mock redis client（记录当前队列内容）。"""
    client = MagicMock()
    store: dict[str, list[bytes]] = {"q": list(payloads)}

    def lrange(name, start, end):  # noqa: ARG001
        return list(store.get(name, []))

    client.lrange.side_effect = lrange

    pipe = MagicMock()
    pipe.delete.side_effect = lambda name: store.pop(name, None)
    pipe.rpush.side_effect = lambda name, *items: store.setdefault(name, []).extend(items)
    client.pipeline.return_value = pipe
    return client, store, pipe


def test_purge_one_queue_removes_by_biz_id_and_keeps_order():
    stale = _envelope("c-stale", [["biz-stale"], {}, {}])
    keep1 = _envelope("c-keep-1", [["biz-keep-1"], {}, {}])
    keep2 = _envelope("c-keep-2", [["biz-keep-2"], {}, {}])
    client, store, pipe = _queue_with_messages(keep1, stale, keep2)

    removed = guard._purge_one_queue(client, "q", {"biz-stale"}, set())
    assert removed == 1
    # 队列被 delete + rpush 重写，未命中消息保持相对顺序
    pipe.delete.assert_called_once_with("q")
    pipe.rpush.assert_called_once()
    args, _ = pipe.rpush.call_args
    assert args[0] == "q"
    assert list(args[1:]) == [keep1, keep2]
    pipe.execute.assert_called_once()


def test_purge_one_queue_removes_by_celery_message_id():
    stale = _envelope("celery-xyz", [["biz-whatever"], {}, {}])
    keep = _envelope("celery-aaa", [["biz-keep"], {}, {}])
    client, store, pipe = _queue_with_messages(stale, keep)

    removed = guard._purge_one_queue(client, "q", set(), {"celery-xyz"})
    assert removed == 1
    args, _ = pipe.rpush.call_args
    assert list(args[1:]) == [keep]


def test_purge_one_queue_no_hit_no_rewrite():
    msg1 = _envelope("c1", [["b1"], {}, {}])
    msg2 = _envelope("c2", [["b2"], {}, {}])
    client, store, pipe = _queue_with_messages(msg1, msg2)

    removed = guard._purge_one_queue(client, "q", {"other"}, {"other-c"})
    assert removed == 0
    # 没有命中：不重写队列
    pipe.delete.assert_not_called()
    pipe.rpush.assert_not_called()


def test_purge_one_queue_all_removed_deletes_without_rpush():
    stale = _envelope("c-stale", [["biz-stale"], {}, {}])
    client, store, pipe = _queue_with_messages(stale)

    removed = guard._purge_one_queue(client, "q", {"biz-stale"}, set())
    assert removed == 1
    pipe.delete.assert_called_once_with("q")
    pipe.rpush.assert_not_called()


def test_purge_one_queue_lrange_exception_returns_zero():
    client = MagicMock()
    client.lrange.side_effect = RuntimeError("redis down")
    assert guard._purge_one_queue(client, "q", {"b"}, set()) == 0


def test_purge_one_queue_empty_queue_returns_zero():
    client = MagicMock()
    client.lrange.return_value = []
    assert guard._purge_one_queue(client, "q", {"b"}, set()) == 0
    client.pipeline.assert_not_called()


def test_purge_one_queue_rewrite_exception_returns_zero():
    stale = _envelope("c-stale", [["biz-stale"], {}, {}])
    client, store, pipe = _queue_with_messages(stale)
    pipe.execute.side_effect = RuntimeError("write fail")

    removed = guard._purge_one_queue(client, "q", {"biz-stale"}, set())
    assert removed == 0


def test_purge_one_queue_unparseable_message_conservatively_kept():
    stale = _envelope("c-stale", [["biz-stale"], {}, {}])
    garbage = b"garbage-not-a-message"
    client, store, pipe = _queue_with_messages(garbage, stale)

    removed = guard._purge_one_queue(client, "q", {"biz-stale"}, set())
    assert removed == 1
    args, _ = pipe.rpush.call_args
    # 无法解析的消息保守保留，绝不误删
    assert list(args[1:]) == [garbage]


# ── purge_stale_messages_from_queues ────────────────────────────────────


def test_purge_queues_no_ids_returns_zero_without_connecting():
    assert guard.purge_stale_messages_from_queues("redis://x", ("q",)) == 0


def test_purge_queues_blank_ids_filtered_out():
    assert guard.purge_stale_messages_from_queues("redis://x", ("q",), business_task_ids=["", None]) == 0


def test_purge_queues_redis_not_installed(monkeypatch):
    """redis-py 不可用（ImportError）时安全返回 0。"""
    import builtins

    real_import = builtins.__import__

    def fake_import(name, *args, **kwargs):
        if name == "redis":
            raise ImportError("no redis")
        return real_import(name, *args, **kwargs)

    monkeypatch.setattr(builtins, "__import__", fake_import)
    assert guard.purge_stale_messages_from_queues("redis://x", ("q",), business_task_ids=["b1"]) == 0


def test_purge_queues_connection_failure_returns_zero():
    fake_redis = types.ModuleType("redis")

    class _FakeRedis:
        @classmethod
        def from_url(cls, url):  # noqa: ARG003
            client = MagicMock()
            client.ping.side_effect = ConnectionError("connect refused")
            return client

    fake_redis.Redis = _FakeRedis
    sys.modules["redis"] = fake_redis
    try:
        assert guard.purge_stale_messages_from_queues("redis://x", ("q",), celery_task_ids=["c1"]) == 0
    finally:
        sys.modules.pop("redis", None)


def test_purge_queues_happy_path_closes_client():
    stale = _envelope("c-stale", [["biz-stale"], {}, {}])
    fake_redis = types.ModuleType("redis")

    client = MagicMock()
    client.lrange.return_value = [stale]
    pipe = MagicMock()
    client.pipeline.return_value = pipe

    class _FakeRedis:
        @classmethod
        def from_url(cls, url):  # noqa: ARG003
            return client

    fake_redis.Redis = _FakeRedis
    sys.modules["redis"] = fake_redis
    try:
        removed = guard.purge_stale_messages_from_queues(
            "redis://x", ("generation", "transcode"), business_task_ids=["biz-stale"]
        )
    finally:
        sys.modules.pop("redis", None)

    # mock client 对两个队列都返回同一条作废消息 → 各移除 1 条
    assert removed == 2
    client.ping.assert_called_once()
    client.close.assert_called_once()
    # 两个队列都扫描
    assert client.lrange.call_count == 2


def test_purge_queues_close_exception_swallowed():
    fake_redis = types.ModuleType("redis")

    client = MagicMock()
    client.lrange.return_value = []
    client.close.side_effect = RuntimeError("close fail")

    class _FakeRedis:
        @classmethod
        def from_url(cls, url):  # noqa: ARG003
            return client

    fake_redis.Redis = _FakeRedis
    sys.modules["redis"] = fake_redis
    try:
        removed = guard.purge_stale_messages_from_queues("redis://x", ("q",), celery_task_ids=["c1"])
    finally:
        sys.modules.pop("redis", None)
    assert removed == 0


# ── revoke_and_purge ────────────────────────────────────────────────────


def test_revoke_and_purge_revokes_each_message(monkeypatch):
    fake_app = MagicMock()
    purge_mock = MagicMock(return_value=2)
    monkeypatch.setattr(guard, "purge_stale_messages_from_queues", purge_mock)

    removed = guard.revoke_and_purge(
        fake_app,
        "redis://x",
        business_task_ids=["b1"],
        celery_task_ids=["c1", "c2"],
        queue_names=("generation",),
    )
    assert removed == 2
    assert fake_app.control.revoke.call_count == 2
    fake_app.control.revoke.assert_any_call("c1")
    fake_app.control.revoke.assert_any_call("c2")
    purge_mock.assert_called_once_with(
        "redis://x", ("generation",), business_task_ids=["b1"], celery_task_ids=["c1", "c2"]
    )


def test_revoke_and_purge_revoke_exception_does_not_block(monkeypatch):
    fake_app = MagicMock()
    fake_app.control.revoke.side_effect = RuntimeError("broadcast fail")
    purge_mock = MagicMock(return_value=0)
    monkeypatch.setattr(guard, "purge_stale_messages_from_queues", purge_mock)

    removed = guard.revoke_and_purge(fake_app, "redis://x", celery_task_ids=["c1"])
    assert removed == 0
    purge_mock.assert_called_once()


def test_revoke_and_purge_skips_blank_ids(monkeypatch):
    fake_app = MagicMock()
    purge_mock = MagicMock(return_value=0)
    monkeypatch.setattr(guard, "purge_stale_messages_from_queues", purge_mock)

    guard.revoke_and_purge(fake_app, "redis://x", celery_task_ids=["", None])
    fake_app.control.revoke.assert_not_called()


# ── ensure_task_claimable ───────────────────────────────────────────────


def test_ensure_claimable_missing_task_returns_empty():
    assert guard.ensure_task_claimable("t1", lambda _tid: None) == ""


def test_ensure_claimable_terminal_raises():
    with pytest.raises(guard.StaleTaskDiscarded) as exc_info:
        guard.ensure_task_claimable("t1", lambda _tid: "failed")
    assert exc_info.value.task_id == "t1"
    assert exc_info.value.status == "failed"


def test_ensure_claimable_cancelled_raises():
    with pytest.raises(guard.StaleTaskDiscarded):
        guard.ensure_task_claimable("t1", lambda _tid: "cancelled")


def test_ensure_claimable_pending_passes():
    assert guard.ensure_task_claimable("t1", lambda _tid: "pending") == "pending"
