# -*- encoding=utf8 -*-
"""
Block 测试任务功能单元测试

覆盖:
1. Block 任务加载（task_key / task_details / 高优先队列）
2. Worker 收到的 payload 包含 blocked 参数
3. Block 任务重试逻辑（下载错误重试，其他错误直接失败）
4. Analytics snapshot 跳过 block 任务
5. Block 测试报告生成
"""

import json
import sys
import os
import tempfile
import csv

PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, PROJECT_ROOT)

import pytest

from analytics import _make_task_key


class FakeRedis:
    """模拟 Redis 操作"""

    def __init__(self):
        self.lists: dict = {}
        self.hashes: dict = {}
        self.sets: dict = {}
        self.published: list = []

    def delete(self, key):
        self.lists.pop(key, None)
        self.hashes.pop(key, None)
        self.sets.pop(key, None)

    def lpush(self, key, value):
        self.lists.setdefault(key, []).insert(0, value)

    def rpush(self, key, value):
        self.lists.setdefault(key, []).append(value)

    def rpop(self, key):
        values = self.lists.setdefault(key, [])
        return values.pop() if values else None

    def lrange(self, key, start, end):
        values = self.lists.get(key, [])
        stop = None if end == -1 else end + 1
        return values[start:stop]

    def llen(self, key):
        return len(self.lists.get(key, []))

    def lpos(self, key, value):
        try:
            return self.lists.get(key, []).index(value)
        except ValueError:
            return None

    def lrem(self, key, count, value):
        values = self.lists.get(key, [])
        original_len = len(values)
        self.lists[key] = [item for item in values if item != value]
        return original_len - len(self.lists[key])

    def hset(self, key, *args, **kwargs):
        if len(args) == 2:
            field, value = args
            self.hashes.setdefault(key, {})[field] = value
        elif len(args) == 1 and isinstance(args[0], dict):
            self.hashes.setdefault(key, {}).update(args[0])

    def hget(self, key, field):
        return self.hashes.get(key, {}).get(field)

    def hgetall(self, key):
        return dict(self.hashes.get(key, {}))

    def hvals(self, key):
        return list(self.hashes.get(key, {}).values())

    def hlen(self, key):
        return len(self.hashes.get(key, {}))

    def hexists(self, key, field):
        return field in self.hashes.get(key, {})

    def hdel(self, key, field):
        self.hashes.get(key, {}).pop(field, None)

    def sadd(self, key, value):
        self.sets.setdefault(key, set()).add(value)

    def srem(self, key, value):
        self.sets.setdefault(key, set()).discard(value)

    def scard(self, key):
        return len(self.sets.get(key, set()))

    def smembers(self, key):
        return set(self.sets.get(key, set()))

    def sismember(self, key, value):
        return value in self.sets.get(key, set())

    def exists(self, key):
        return int(key in self.hashes or key in self.lists or key in self.sets)

    def publish(self, channel, message):
        self.published.append((channel, message))

    def keys(self, pattern="*"):
        result = []
        for name in list(self.lists) + list(self.hashes) + list(self.sets):
            if pattern == "*" or name.startswith(pattern.replace("*", "")):
                result.append(name)
        return result

    def scan_iter(self, match="*"):
        for key in self.keys(match):
            yield key

    def setex(self, key, ttl, value):
        pass

    def pipeline(self, transaction=True):
        return _FakePipeline(self)

    @staticmethod
    def lock(lock_key, timeout=5, blocking_timeout=1):
        class _FakeLock:
            def __enter__(self2):
                pass

            def __exit__(self2, *args):
                pass

        return _FakeLock()


class _FakePipeline:
    """模拟 Redis pipeline"""

    def __init__(self, redis_conn):
        self._redis = redis_conn

    def __enter__(self):
        return self

    def __exit__(self, *args):
        pass

    def sadd(self, key, value):
        self._redis.sadd(key, value)
        return self

    def srem(self, key, value):
        self._redis.srem(key, value)
        return self

    def hset(self, key, field, value):
        self._redis.hset(key, field, value)
        return self

    def hdel(self, key, field):
        self._redis.hdel(key, field)
        return self

    def delete(self, key):
        self._redis.delete(key)
        return self

    def expire(self, key, ttl):
        return self

    def execute(self):
        pass


class StubAnalytics:
    """模拟 Analytics 服务"""

    def __init__(self, pending=None, model=None, qualified=None):
        self.pending = pending or []
        self.model = model or []
        self.qualified = qualified or []
        self.enqueued_incremental = []

    def list_pending_collection_tasks(self):
        return list(self.pending)

    def list_model_eligible_apps(self):
        return list(self.model)

    def enqueue_incremental(self, package_name, **kwargs):
        self.enqueued_incremental.append((package_name, kwargs))


class StubMonitor:
    """模拟 Monitoring 服务"""

    def __init__(self):
        self.failure_buckets: dict = {}
        self.assigned_tasks: list = []
        self.reports: list = []

    def set_failure_bucket(self, task_key, bucket):
        self.failure_buckets[task_key] = bucket

    def handle_task_assigned(self, worker_id, task_key, **kwargs):
        self.assigned_tasks.append((worker_id, task_key))

    def handle_worker_report(self, **kwargs):
        self.reports.append(kwargs)

    def set_worker_state(self, worker_id, state, **kwargs):
        pass

    def register_worker(self, worker_id, **kwargs):
        pass

    def mark_worker_offline(self, worker_id, reason):
        pass

    def close_controller_session(self, **kwargs):
        pass

    def start_controller_session(self, **kwargs):
        pass

    def heartbeat_controller_session(self):
        pass


class StubNotifier:
    def send_weCom_alert(self, *args, **kwargs):
        pass

    def send_weChat_alert(self, *args, **kwargs):
        pass


class StubAnalyticsRepo:
    def get_collection_row(self, package_name):
        return None

    def get_latest_task_execution(self, package_name):
        return None

    def has_successful_related_magic_label(self, app_magic_label, package_name):
        return False

    def list_qualified_apps(self):
        return list(self.qualified_apps) if hasattr(self, 'qualified_apps') else []


class StubAnalyticsWithRepo:
    def __init__(self, qualified_apps=None):
        self.repo = StubAnalyticsRepo()
        if qualified_apps:
            self.repo.qualified_apps = qualified_apps
        self.enqueued_incremental = []

    def list_pending_collection_tasks(self):
        return []

    def list_model_eligible_apps(self):
        return []

    def list_qualified_apps(self):
        return self.repo.list_qualified_apps()

    def enqueue_incremental(self, **kwargs):
        self.enqueued_incremental.append(kwargs)

    def close(self):
        pass


def make_dispatcher(redis_conn=None, analytics=None, monitor=None, notifier=None):
    """构建一个最小化的 RedisTaskDispatcher 用于测试"""
    from redis_task_distribute import RedisTaskDispatcher

    dispatcher = RedisTaskDispatcher.__new__(RedisTaskDispatcher)
    dispatcher.redis = redis_conn or FakeRedis()
    dispatcher.analytics = analytics or StubAnalytics()
    dispatcher.monitor = monitor or StubMonitor()
    dispatcher.notifier = notifier or StubNotifier()
    dispatcher.worker_inventory = {}
    dispatcher.managed_worker_ids = set()
    dispatcher.only_managed_workers_can_dispatch = False
    dispatcher.worker_online_timeout = 300
    dispatcher.task_routing_rules = {"package_name": {}, "task_key": {}}
    dispatcher._apk_registry = None
    dispatcher._minio_storage = None

    import types
    def _mark_noop(self, package_name, app_name="", last_updated=""):
        return None
    dispatcher._mark_download_error_awaiting_apk = types.MethodType(_mark_noop, dispatcher)

    def _dispatch_alert_noop(self, *args, **kwargs):
        pass
    dispatcher._dispatch_alert = types.MethodType(_dispatch_alert_noop, dispatcher)

    def _registry_is_fresh_enough_noop(package_name, last_updated):
        return False
    class _StubRegistry:
        is_fresh_enough = staticmethod(_registry_is_fresh_enough_noop)
    dispatcher._get_apk_registry = lambda: _StubRegistry
    return dispatcher


def qualified_entry(app_name, package_name, country_code="US", device_type="emulator"):
    """构建一个合格的 app 条目"""
    return {
        "package_name": package_name,
        "app_name": app_name,
        "task_payload": {
            "app_name": app_name,
            "package_name": package_name,
            "country_code": country_code,
            "device_type": device_type,
            "original_row": {},
        },
    }


# =============================================================
# 1. Block 任务加载测试
# =============================================================


class TestBlockTaskLoading:
    """测试 block 任务的加载行为"""

    def test_block_task_has_correct_task_key_suffix(self):
        """block 任务 key 以 _block 结尾"""
        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        entries = [qualified_entry("TestApp", "com.test.block")]
        count = dispatcher.load_block_tasks(entries)
        assert count == 1

        expected_key = "TestApp_com.test.block_block"
        assert redis_conn.hexists("task:details", expected_key), (
            f"expected task_key '{expected_key}' in task:details"
        )

    def test_block_tasks_pushed_to_high_priority_queue(self):
        """block 任务入高优队列"""
        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        entries = [
            qualified_entry("App1", "com.app1"),
            qualified_entry("App2", "com.app2"),
        ]
        count = dispatcher.load_block_tasks(entries)
        assert count == 2

        high_queue = redis_conn.lrange("task:queue:high", 0, -1)
        assert len(high_queue) == 2
        for task_key in high_queue:
            assert task_key.endswith("_block")

    def test_block_task_details_contain_is_block_task_flag(self):
        """task_details 包含 is_block_task=True"""
        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        entries = [qualified_entry("BlockApp", "com.block.test")]
        dispatcher.load_block_tasks(entries)

        task_key = "BlockApp_com.block.test_block"
        details_json = redis_conn.hget("task:details", task_key)
        details = json.loads(details_json)
        assert details.get("is_block_task") is True
        assert details.get("task_queue") == "high"

    def test_block_task_has_app_name_and_package_in_details(self):
        """task_details 保留原始 app_name 和 package_name"""
        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        entries = [
            qualified_entry("MyApp", "com.company.myapp", country_code="CN", device_type="emulator")
        ]
        dispatcher.load_block_tasks(entries)

        task_key = "MyApp_com.company.myapp_block"
        details = json.loads(redis_conn.hget("task:details", task_key))
        assert details["app_name"] == "MyApp"
        assert details["package_name"] == "com.company.myapp"
        assert details["country_code"] == "CN"
        assert details["device_type"] == "emulator"

    def test_duplicate_block_tasks_are_skipped(self):
        """重复加载 block 任务先清除旧任务再重新加载，每次 count 相同"""
        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        entries = [qualified_entry("DupApp", "com.dup")]
        assert dispatcher.load_block_tasks(entries) == 1
        # 第二次加载先清除旧任务再重新加载，不是简单地跳过
        assert dispatcher.load_block_tasks(entries) == 1
        # 队列中只有一个任务（不是两个）
        high_queue = redis_conn.lrange("task:queue:high", 0, -1)
        assert len(high_queue) == 1

    def test_load_all_qualified_block_tasks_calls_repo(self):
        """load_all_qualified_block_tasks 从 DB 查询 qualified apps"""
        redis_conn = FakeRedis()
        analytics = StubAnalyticsWithRepo(qualified_apps=[
            qualified_entry("QualApp", "com.qual"),
            qualified_entry("AnotherQual", "com.another"),
        ])
        dispatcher = make_dispatcher(redis_conn=redis_conn, analytics=analytics)

        count = dispatcher.load_all_qualified_block_tasks()
        assert count == 2

        assert redis_conn.hexists("task:details", "QualApp_com.qual_block")
        assert redis_conn.hexists("task:details", "AnotherQual_com.another_block")


# =============================================================
# 2. Worker 接收 payload 测试
# =============================================================


class TestBlockTaskPayload:
    """测试 Worker 收到的 task payload"""

    def test_task_payload_includes_traffic_root(self):
        """_task_to_payload 的 traffic_root 和 blocked 字段：block 任务用 block 路径"""
        from config import ANALYTICS_TRAFFIC_ROOT_BLOCK

        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        task_details = {
            "app_name": "TestApp",
            "package_name": "com.test",
            "country_code": "US",
            "is_block_task": True,
            "available_sources": ["google_play", "local"],
        }
        payload = dispatcher._task_to_payload("TestApp_com.test_block", task_details)
        assert payload["blocked"] is True
        assert payload["traffic_root"] == ANALYTICS_TRAFFIC_ROOT_BLOCK

    def test_normal_task_payload_uses_default_traffic_root(self):
        """普通任务的 blocked=False, traffic_root 为默认路径"""
        from config import ANALYTICS_TRAFFIC_ROOT

        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        task_details = {
            "app_name": "NormalApp",
            "package_name": "com.normal",
            "country_code": "US",
            "available_sources": ["google_play", "local"],
        }
        payload = dispatcher._task_to_payload("NormalApp_com.normal", task_details)
        assert payload["blocked"] is False
        assert payload["traffic_root"] == ANALYTICS_TRAFFIC_ROOT

    def test_worker_init_receives_block_task_with_traffic_root(self):
        """Worker init 返回的 task 有正确的 traffic_root"""
        from config import ANALYTICS_TRAFFIC_ROOT_BLOCK
        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        entries = [qualified_entry("BlockInit", "com.block.init")]
        dispatcher.load_block_tasks(entries)

        task_key = "BlockInit_com.block.init_block"

        worker_id = "192.168.1.1_AA:BB:CC:DD:EE:01"
        worker_info = {
            "worker_id": worker_id,
            "ip_address": "192.168.1.1",
            "mac_address": "AA:BB:CC:DD:EE:01",
            "hostname": "test-host",
            "platform": "Windows",
            "device_type": "emulator",
            "register_time": 1000000000.0,
            "last_update": 1000000000.0,
            "status": "idle",
            "dispatch_enabled": True,
        }
        redis_conn.hset("workers:info", worker_id, json.dumps(worker_info))
        redis_conn.sadd("workers:idle", worker_id)

        task = dispatcher.worker_init(
            worker_id=worker_id,
            ip_address="192.168.1.1",
            mac_address="AA:BB:CC:DD:EE:01",
            hostname="test-host",
            platform="Windows",
            device_type="emulator",
        )

        assert task is not None
        assert task["blocked"] is True
        assert task["traffic_root"] == ANALYTICS_TRAFFIC_ROOT_BLOCK
        assert task["task_key"] == task_key

    def test_worker_report_then_assign_next_block_task(self):
        """Worker 上报完成后领取下一个 block 任务"""
        redis_conn = FakeRedis()
        monitor = StubMonitor()
        notifier = StubNotifier()
        dispatcher = make_dispatcher(
            redis_conn=redis_conn, monitor=monitor, notifier=notifier
        )

        entries = [
            qualified_entry("BlockA", "com.block.a"),
            qualified_entry("BlockB", "com.block.b"),
        ]
        dispatcher.load_block_tasks(entries)

        task_a_key = "BlockA_com.block.a_block"
        task_b_key = "BlockB_com.block.b_block"

        worker_id = "192.168.1.1_AA:BB:CC:DD:EE:01"
        worker_info = {
            "worker_id": worker_id,
            "ip_address": "192.168.1.1",
            "mac_address": "AA:BB:CC:DD:EE:01",
            "hostname": "test-host",
            "platform": "Windows",
            "device_type": "emulator",
            "register_time": 1000000000.0,
            "last_update": 1000000000.0,
            "status": "idle",
            "dispatch_enabled": True,
        }
        redis_conn.hset("workers:info", worker_id, json.dumps(worker_info))
        redis_conn.sadd("workers:idle", worker_id)

        first_task = dispatcher.worker_init(
            worker_id=worker_id,
            ip_address="192.168.1.1",
            mac_address="AA:BB:CC:DD:EE:01",
            hostname="test-host",
            platform="Windows",
            device_type="emulator",
        )
        assert first_task is not None

        second_task = dispatcher.worker_report(
            worker_id=worker_id,
            previous_task_key=first_task["task_key"],
            report_data={"status": "success"},
        )
        assert second_task is not None
        from config import ANALYTICS_TRAFFIC_ROOT_BLOCK
        assert second_task["traffic_root"] == ANALYTICS_TRAFFIC_ROOT_BLOCK


# =============================================================
# 3. Block 任务重试逻辑测试
# =============================================================


class TestBlockTaskRetry:
    """测试 block 任务的专用重试策略"""

    def _setup_block_task(self, redis_conn, task_key, app_name="BlockApp", package_name="com.block"):
        """准备一个 block 任务的 Redis 数据"""
        task_details = {
            "app_name": app_name,
            "package_name": package_name,
            "country_code": "US",
            "is_block_task": True,
            "task_queue": "high",
        }
        redis_conn.hset("task:details", task_key, json.dumps(task_details))
        redis_conn.hset("task:status", task_key, json.dumps({
            "status": "running",
            "worker_id": "192.168.1.1_WORKER",
            "retry_count": 0,
        }))

    def test_block_task_download_error_goes_to_pending_retry(self):
        """block 任务的下载错误 -> pending_retry"""
        from result_codes import ErrorInfo, ErrorCategory, DownloadError

        redis_conn = FakeRedis()
        monitor = StubMonitor()
        notifier = StubNotifier()
        dispatcher = make_dispatcher(
            redis_conn=redis_conn, monitor=monitor, notifier=notifier
        )

        task_key = "BlockApp_com.block_block"
        self._setup_block_task(redis_conn, task_key)

        dispatcher._record_failed_task(
            task_key=task_key,
            worker_id="192.168.1.1_WORKER",
            error_info=ErrorInfo.download(DownloadError.REGION_RESTRICTED),
            message="Google Play 锁区",
        )

        status = json.loads(redis_conn.hget("task:status", task_key))
        assert status["status"] == "pending_retry", (
            f"expected pending_retry, got {status['status']}"
        )
        assert monitor.failure_buckets.get(task_key) == "retry"

    def test_block_task_app_error_is_permanent_fail(self):
        """block 任务的 APP_ERROR -> 永久失败"""
        from result_codes import ErrorInfo, ErrorCategory, AppError

        redis_conn = FakeRedis()
        monitor = StubMonitor()
        notifier = StubNotifier()
        dispatcher = make_dispatcher(
            redis_conn=redis_conn, monitor=monitor, notifier=notifier
        )

        task_key = "BlockApp_com.block_block"
        self._setup_block_task(redis_conn, task_key)

        dispatcher._record_failed_task(
            task_key=task_key,
            worker_id="192.168.1.1_WORKER",
            error_info=ErrorInfo.app(AppError.CRASH),
            message="应用闪退",
        )

        status = json.loads(redis_conn.hget("task:status", task_key))
        assert status["status"] == "failed"
        assert redis_conn.sismember("task:failed", task_key) if hasattr(redis_conn, 'sismember') else (
            task_key in redis_conn.smembers("task:failed")
        )
        assert monitor.failure_buckets.get(task_key) == "failed"

    def test_block_task_business_error_is_permanent_fail(self):
        """block 任务的 BUSINESS_ERROR -> 永久失败"""
        from result_codes import ErrorInfo, ErrorCategory, BusinessError

        redis_conn = FakeRedis()
        monitor = StubMonitor()
        notifier = StubNotifier()
        dispatcher = make_dispatcher(
            redis_conn=redis_conn, monitor=monitor, notifier=notifier
        )

        task_key = "BlockApp_com.block_block"
        self._setup_block_task(redis_conn, task_key)

        dispatcher._record_failed_task(
            task_key=task_key,
            worker_id="192.168.1.1_WORKER",
            error_info=ErrorInfo.business(BusinessError.LOGIN_FAILED),
            message="登录失败",
        )

        status = json.loads(redis_conn.hget("task:status", task_key))
        assert status["status"] == "failed"
        assert task_key in redis_conn.smembers("task:failed")

    def test_block_task_infra_non_download_error_is_permanent_fail(self):
        """block 任务的非下载 INFRA_ERROR -> 永久失败"""
        from result_codes import ErrorInfo, ErrorCategory, InfraError

        redis_conn = FakeRedis()
        monitor = StubMonitor()
        notifier = StubNotifier()
        dispatcher = make_dispatcher(
            redis_conn=redis_conn, monitor=monitor, notifier=notifier
        )

        task_key = "BlockApp_com.block_block"
        self._setup_block_task(redis_conn, task_key)

        dispatcher._record_failed_task(
            task_key=task_key,
            worker_id="192.168.1.1_WORKER",
            error_info=ErrorInfo.infra(InfraError.ADB_ERROR),
            message="ADB 断联",
        )

        status = json.loads(redis_conn.hget("task:status", task_key))
        assert status["status"] == "failed"
        assert task_key in redis_conn.smembers("task:failed")

    def test_block_task_infra_download_failed_goes_to_retry(self):
        """block 任务 INFRA_ERROR.DOWNLOAD_FAILED -> pending_retry"""
        from result_codes import ErrorInfo, ErrorCategory, InfraError

        redis_conn = FakeRedis()
        monitor = StubMonitor()
        notifier = StubNotifier()
        dispatcher = make_dispatcher(
            redis_conn=redis_conn, monitor=monitor, notifier=notifier
        )

        task_key = "BlockApp_com.block_block"
        self._setup_block_task(redis_conn, task_key)

        dispatcher._record_failed_task(
            task_key=task_key,
            worker_id="192.168.1.1_WORKER",
            error_info=ErrorInfo.infra(InfraError.DOWNLOAD_FAILED),
            message="下载失败",
        )

        status = json.loads(redis_conn.hget("task:status", task_key))
        assert status["status"] == "pending_retry", (
            f"expected pending_retry for DOWNLOAD_FAILED, got {status['status']}"
        )
        assert monitor.failure_buckets.get(task_key) == "retry"

    def test_block_task_retry_avoids_same_worker(self):
        """block 任务下载失败重试时避开同一 worker"""
        from result_codes import ErrorInfo, ErrorCategory, DownloadError

        redis_conn = FakeRedis()
        monitor = StubMonitor()
        notifier = StubNotifier()
        dispatcher = make_dispatcher(
            redis_conn=redis_conn, monitor=monitor, notifier=notifier
        )

        task_key = "BlockApp_com.block_block"
        worker_id = "192.168.2.1_FAILED_WORKER"
        self._setup_block_task(redis_conn, task_key)

        dispatcher._record_failed_task(
            task_key=task_key,
            worker_id=worker_id,
            error_info=ErrorInfo.download(DownloadError.APP_NOT_FOUND),
            message="Google Play 未找到",
        )

        details = json.loads(redis_conn.hget("task:details", task_key))
        assert worker_id in details.get("excluded_worker_ids", [])

    def test_normal_task_retry_unchanged(self):
        """block 任务的逻辑不影响普通任务重试"""
        from result_codes import ErrorInfo, ErrorCategory, AppError

        redis_conn = FakeRedis()
        monitor = StubMonitor()
        notifier = StubNotifier()
        dispatcher = make_dispatcher(
            redis_conn=redis_conn, monitor=monitor, notifier=notifier
        )

        task_key = "NormalApp_com.normal"
        redis_conn.hset("task:details", task_key, json.dumps({
            "app_name": "NormalApp",
            "package_name": "com.normal",
            "country_code": "US",
            "device_type": "physical",
        }))
        redis_conn.hset("task:status", task_key, json.dumps({
            "status": "running",
            "worker_id": "worker_x",
            "retry_count": 0,
        }))

        dispatcher._record_failed_task(
            task_key=task_key,
            worker_id="worker_x",
            error_info=ErrorInfo.app(AppError.CRASH),
            message="闪退",
        )

        status = json.loads(redis_conn.hget("task:status", task_key))
        assert status["status"] == "failed"
        assert task_key in redis_conn.smembers("task:failed")


# =============================================================
# 4. Analytics snapshot 跳过测试
# =============================================================


class TestAnalyticsSnapshotSkip:
    """测试 analytics snapshot 回调跳过 block 任务"""

    def _setup_task(self, redis_conn, task_key, is_blocked=False):
        redis_conn.hset("task:details", task_key, json.dumps({
            "app_name": "Test",
            "package_name": "com.test",
            "is_block_task": is_blocked,
            "task_queue": "high",
        }))
        redis_conn.hset("task:status", task_key, json.dumps({
            "status": "pending_retry",
            "retry_count": 1,
        }))

    def test_block_task_skipped_in_analytics_snapshot(self):
        """block 任务在 analytics 回调中被跳过（不重新入队）"""
        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        task_key = "BlockApp_com.block_block"
        self._setup_task(redis_conn, task_key, is_blocked=True)

        summary = {
            "task_payload": {"app_name": "BlockApp", "package_name": "com.block"},
            "latest_task_key": task_key,
            "collection_status": "pending",
        }

        dispatcher._handle_analytics_snapshot("com.block", summary)

        status = json.loads(redis_conn.hget("task:status", task_key))
        assert status["status"] == "pending_retry", (
            f"block task should not be re-queued, got {status['status']}"
        )

    def test_block_task_suffix_also_skipped(self):
        """以 _block 结尾的任务同样被 analytics 跳过"""
        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        task_key = "SomeApp_com.some_block"
        redis_conn.hset("task:details", task_key, json.dumps({
            "app_name": "SomeApp",
            "package_name": "com.some",
        }))

        summary = {
            "latest_task_key": task_key,
            "collection_status": "pending",
        }

        dispatcher._handle_analytics_snapshot("com.some", summary)

        assert not redis_conn.hexists("task:status", task_key), (
            "block-suffix task should be skipped entirely (status not set)"
        )

    def test_normal_task_still_processed_by_analytics(self):
        """普通任务在 analytics 回调中照常处理"""
        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        task_key = "NormalApp_com.normal"
        self._setup_task(redis_conn, task_key, is_blocked=False)

        summary = {
            "task_payload": {"app_name": "NormalApp", "package_name": "com.normal"},
            "latest_task_key": task_key,
            "collection_status": "qualified",
        }

        dispatcher._handle_analytics_snapshot("com.normal", summary)

        assert task_key in redis_conn.smembers("task:completed")

    def test_block_task_redis_state_unchanged_by_callback(self):
        """block 任务无论 collection_status 如何 Redis 状态不变"""
        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        task_key = "BlockApp_com.block_block"
        self._setup_task(redis_conn, task_key, is_blocked=True)

        for status in ["qualified", "failed_terminal", "pending"]:
            summary = {
                "task_payload": {"app_name": "BlockApp", "package_name": "com.block"},
                "latest_task_key": task_key,
                "collection_status": status,
            }
            dispatcher._handle_analytics_snapshot("com.block", summary)

            task_status = json.loads(redis_conn.hget("task:status", task_key))
            assert task_status["status"] == "pending_retry", (
                f"status should stay pending_retry (not affected by analytics), "
                f"got {task_status['status']} for collection_status={status}"
            )


# =============================================================
# 5. Block 测试报告生成测试
# =============================================================


class TestBlockReport:
    """测试 block 测试报告 CSV 生成"""

    def test_report_includes_block_tasks_only(self, tmp_path):
        """报告只包含 _block 后缀的任务"""
        report_path = os.path.join(str(tmp_path), "block_test_report.csv")

        from dispatcher_main import generate_block_test_report

        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        for app_name, package_name, suffix in (
            ("BlockedA", "com.block.a", "_block"),
            ("BlockedB", "com.block.b", "_block"),
            ("NormalApp", "com.normal", ""),
        ):
            task_key = f"{app_name}_{package_name}{suffix}"
            redis_conn.hset("task:details", task_key, json.dumps({
                "app_name": app_name,
                "package_name": package_name,
                "is_block_task": suffix == "_block",
            }))
            redis_conn.hset("task:status", task_key, json.dumps({
                "status": "success" if suffix == "_block" else "completed",
                "worker_id": "worker_x",
                "start_time": 1000.0,
                "end_time": 1100.0,
                "retry_count": 0,
            }))
            redis_conn.hset("workers:info", "worker_x", json.dumps({
                "ip_address": "192.168.1.1",
            }))

        generate_block_test_report(dispatcher, report_dir=str(tmp_path))

        assert os.path.exists(report_path)
        with open(report_path, "r", encoding="utf-8") as f:
            reader = csv.reader(f)
            rows = list(reader)

        assert len(rows) >= 3
        headers = rows[0]
        assert "任务键" in headers
        assert "应用名称" in headers
        assert "状态" in headers
        assert "失败原因" in headers

        task_keys_in_report = [row[0] for row in rows[1:] if row and row[0]]
        assert "BlockedA_com.block.a_block" in task_keys_in_report
        assert "BlockedB_com.block.b_block" in task_keys_in_report
        assert "NormalApp_com.normal" not in task_keys_in_report

    def test_report_counts_success_and_failure(self, tmp_path):
        """报告正确统计成功/失败/重试数量"""
        report_path = os.path.join(str(tmp_path), "block_test_report.csv")

        from dispatcher_main import generate_block_test_report

        redis_conn = FakeRedis()
        dispatcher = make_dispatcher(redis_conn=redis_conn)

        block_data = [
            ("SuccessApp", "com.success", "success"),
            ("FailedApp", "com.failed", "failed"),
            ("RetryingApp", "com.retry", "pending_retry"),
            ("PendingApp", "com.pending", "pending"),
        ]
        for app_name, pkg, status in block_data:
            task_key = f"{app_name}_{pkg}_block"
            redis_conn.hset("task:details", task_key, json.dumps({
                "app_name": app_name,
                "package_name": pkg,
                "is_block_task": True,
            }))
            fail_message = "App crashed" if status == "failed" else ""
            redis_conn.hset("task:status", task_key, json.dumps({
                "status": status,
                "worker_id": "worker_x",
                "last_fail_message": fail_message,
                "retry_count": 0,
            }))

        generate_block_test_report(dispatcher, report_dir=str(tmp_path))

        with open(report_path, "r", encoding="utf-8") as f:
            content = f.read()

        assert "成功,1" in content
        assert "失败,1" in content
        assert "重试中/待分发,2" in content


# =============================================================
# 6. 集成测试: 模拟 Worker 完整接收 block 任务流程
# =============================================================


class TestWorkerBlockTaskIntegration:
    """模拟 Worker 完整的 block 任务接收-执行-上报 链路"""

    def test_worker_receives_block_task_params(self):
        """Worker 通过 init 拿到的 block 任务包含所有预期字段"""
        redis_conn = FakeRedis()
        monitor = StubMonitor()
        dispatcher = make_dispatcher(redis_conn=redis_conn, monitor=monitor)

        entries = [qualified_entry("FullBlock", "com.full.block", country_code="CN", device_type="emulator")]
        dispatcher.load_block_tasks(entries)
        task_key = "FullBlock_com.full.block_block"

        worker_id = "192.168.1.100_EE:FF:00:11:22:33"
        redis_conn.hset("workers:info", worker_id, json.dumps({
            "worker_id": worker_id,
            "ip_address": "192.168.1.100",
            "mac_address": "EE:FF:00:11:22:33",
            "hostname": "block-test",
            "platform": "Windows",
            "device_type": "emulator",
            "register_time": 1000000000.0,
            "last_update": 1000000000.0,
            "status": "idle",
            "dispatch_enabled": True,
        }))
        redis_conn.sadd("workers:idle", worker_id)

        task = dispatcher.worker_init(
            worker_id=worker_id,
            ip_address="192.168.1.100",
            mac_address="EE:FF:00:11:22:33",
            hostname="block-test",
            platform="Windows",
            device_type="emulator",
        )

        assert task is not None
        from config import ANALYTICS_TRAFFIC_ROOT_BLOCK
        assert task["traffic_root"] == ANALYTICS_TRAFFIC_ROOT_BLOCK
        assert task["task_key"] == task_key
        assert task["app_name"] == "FullBlock"
        assert task["package_name"] == "com.full.block"
        assert task["country_code"] == "CN"
        assert task["device_type"] == "emulator"

    def test_full_lifecycle_block_task_to_failure(self):
        """Block 任务完整生命周期: init → report failure → check not re-queued"""
        redis_conn = FakeRedis()
        monitor = StubMonitor()
        notifier = StubNotifier()
        dispatcher = make_dispatcher(
            redis_conn=redis_conn, monitor=monitor, notifier=notifier
        )

        entries = [qualified_entry("LifecycleApp", "com.lifecycle")]
        dispatcher.load_block_tasks(entries)
        task_key = "LifecycleApp_com.lifecycle_block"

        worker_id = "192.168.1.200_AA:11:BB:22:CC:33"
        redis_conn.hset("workers:info", worker_id, json.dumps({
            "worker_id": worker_id,
            "ip_address": "192.168.1.200",
            "mac_address": "AA:11:BB:22:CC:33",
            "hostname": "lifecycle-host",
            "platform": "Windows",
            "device_type": "emulator",
            "register_time": 1000000000.0,
            "last_update": 1000000000.0,
            "status": "idle",
            "dispatch_enabled": True,
        }))
        redis_conn.sadd("workers:idle", worker_id)

        task = dispatcher.worker_init(
            worker_id=worker_id,
            ip_address="192.168.1.200",
            mac_address="AA:11:BB:22:CC:33",
            hostname="lifecycle-host",
            platform="Windows",
            device_type="emulator",
        )
        assert task is not None

        next_task = dispatcher.worker_report(
            worker_id=worker_id,
            previous_task_key=task_key,
            report_data={
                "status": "failed",
                "error": {
                    "category": "APP_ERROR",
                    "code": 1,
                    "reason": "应用闪退",
                },
            },
        )

        status = json.loads(redis_conn.hget("task:status", task_key))
        assert status["status"] == "failed"
        assert task_key in redis_conn.smembers("task:failed")

        assert next_task is None, (
            f"after block task fails, worker should get no next task (queue empty), "
            f"got {next_task}"
        )

    def test_block_task_download_failure_still_in_pending_retry(self):
        """Block 任务下载失败后状态为 pending_retry，等待 APK"""
        redis_conn = FakeRedis()
        monitor = StubMonitor()
        notifier = StubNotifier()
        dispatcher = make_dispatcher(
            redis_conn=redis_conn, monitor=monitor, notifier=notifier
        )

        entries = [qualified_entry("DownloadFail", "com.dl.fail")]
        dispatcher.load_block_tasks(entries)
        task_key = "DownloadFail_com.dl.fail_block"

        worker_id = "192.168.1.201_WORKER"
        redis_conn.hset("workers:info", worker_id, json.dumps({
            "worker_id": worker_id,
            "ip_address": "192.168.1.201",
            "mac_address": "EE:FF:GG:HH:II:JJ",
            "hostname": "dl-fail-host",
            "platform": "Windows",
            "device_type": "emulator",
            "register_time": 1000000000.0,
            "last_update": 1000000000.0,
            "status": "idle",
            "dispatch_enabled": True,
        }))
        redis_conn.sadd("workers:idle", worker_id)

        task = dispatcher.worker_init(
            worker_id=worker_id,
            ip_address="192.168.1.201",
            mac_address="EE:FF:GG:HH:II:JJ",
            hostname="dl-fail-host",
            platform="Windows",
            device_type="emulator",
        )
        assert task is not None

        next_task = dispatcher.worker_report(
            worker_id=worker_id,
            previous_task_key=task_key,
            report_data={
                "status": "failed",
                "error": {
                    "category": "DOWNLOAD_ERROR",
                    "code": 1,
                    "reason": "Google Play 锁区",
                },
            },
        )

        status = json.loads(redis_conn.hget("task:status", task_key))
        assert status["status"] == "pending_retry", (
            f"download error should leave task in pending_retry, got {status['status']}"
        )
        assert task_key not in redis_conn.smembers("task:failed")