"""
app_download_record 表 + ApkRegistry 新增方法 + dispatcher 集成的完整测试。

用法:
    uv run python -m pytest tests/test_download_record.py -v
    uv run python -m pytest tests/test_download_record.py -v -k "TestMinio"  # 仅 MinIO 测试
"""

import json
import os
import shutil
import sqlite3
import sys
import tempfile
import threading
import time
from datetime import datetime, timedelta
from typing import Any, Dict, List
from zoneinfo import ZoneInfo

import pytest

PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if PROJECT_ROOT not in sys.path:
    sys.path.insert(0, PROJECT_ROOT)

from apk_cloud.registry import ApkRegistry, _is_apk_stale, _parse_date, _last_updated_ts

try:
    from apk_cloud.storage import MinioStorage
    _HAS_MINIO = True
except Exception:
    _HAS_MINIO = False

try:
    from redis_task_distribute import RedisTaskDispatcher
    _HAS_DISPATCHER = True
except Exception:
    _HAS_DISPATCHER = False


# ── helpers ──────────────────────────────────────────────────────────

@pytest.fixture
def tmp_db_path():
    fd, path = tempfile.mkstemp(suffix=".sqlite3")
    os.close(fd)
    yield path
    try:
        os.unlink(path)
    except OSError:
        pass


@pytest.fixture
def tmp_storage_dir():
    d = tempfile.mkdtemp(prefix="apk_test_")
    yield d
    shutil.rmtree(d, ignore_errors=True)


@pytest.fixture
def registry(tmp_db_path, tmp_storage_dir):
    r = ApkRegistry(db_path=tmp_db_path, storage_dir=tmp_storage_dir)
    yield r
    try:
        os.unlink(tmp_db_path)
    except OSError:
        pass


@pytest.fixture(autouse=True)
def _cleanup_download_records(registry):
    """每个测试后清理 download_record，确保测试隔离。"""
    yield
    with registry._connect() as conn:
        conn.execute("DELETE FROM app_download_record")


def _make_result(status: str = "ok", sources: Dict[str, Dict[str, Any]] = None,
                 files: List[Dict[str, Any]] = None, download_date: str = "",
                 version_name: str = "") -> Dict[str, Any]:
    files = files if files is not None else [{"filename": "base.apk", "size": 5000000}]
    return {
        "status": status,
        "download_date": download_date or datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "version_name": version_name,
        "files": files,
        "source_details": sources or {},
    }


# ══════════════════════════════════════════════════════════════════════
# 1. Schema & 表结构
# ══════════════════════════════════════════════════════════════════════

class TestSchema:
    def test_table_exists(self, registry):
        with registry._connect() as conn:
            row = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='app_download_record'"
            ).fetchone()
        assert row is not None, "app_download_record table should exist"

    def test_table_columns(self, registry):
        expected = {
            "id", "package_name", "country_code", "app_name", "last_updated",
            "downloads", "worker_play_status", "worker_play_error", "worker_play_date",
            "us_play_status", "us_play_error", "us_play_date",
            "us_aurora_status", "us_aurora_error", "us_aurora_date",
            "apkpure_status", "apkpure_error", "apkpure_date",
            "overall_status", "overall_error", "apk_version_name",
            "local_apk_dir", "local_apk_file_count", "local_apk_is_stale",
            "created_at", "updated_at",
        }
        with registry._connect() as conn:
            cols = {row[1] for row in conn.execute("PRAGMA table_info(app_download_record)")}
        assert expected == cols, f"Missing: {expected - cols}, Extra: {cols - expected}"

    def test_unique_constraint(self, registry):
        now = time.time()
        registry.ensure_download_record("com.test", "US", "App", "2026-05-20", 100)
        with pytest.raises(sqlite3.IntegrityError):
            with registry._connect() as conn:
                conn.execute(
                    "INSERT INTO app_download_record (package_name, country_code, "
                    "last_updated, created_at, updated_at) VALUES (?, ?, ?, ?, ?)",
                    ("com.test", "US", "2026-05-20", now, now),
                )

    def test_default_values(self, registry):
        registry.ensure_download_record("com.test", "US")
        record = registry.get_download_record("com.test", "US")
        assert record["worker_play_status"] == "pending"
        assert record["us_play_status"] == "pending"
        assert record["us_aurora_status"] == "pending"
        assert record["overall_status"] == "pending"
        assert record["local_apk_is_stale"] == 1
        assert record["downloads"] == 0
        assert record["local_apk_file_count"] == 0

    def test_indexes_exist(self, registry):
        with registry._connect() as conn:
            rows = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='index' "
                "AND tbl_name='app_download_record'"
            ).fetchall()
            indexes = {row["name"] for row in rows}
        expected = {
            "idx_download_record_pkg",
            "idx_download_record_country",
            "idx_download_record_overall",
            "idx_download_record_stale",
        }
        assert expected.issubset(indexes), f"Missing indexes: {expected - indexes}"

    def test_create_and_drop_does_not_break(self, registry):
        """验证 DROP + CREATE 不会破坏其他表的数据。"""
        registry.mark_available("com.test.keep", "2026-05-19", "/tmp/dpi/mumu_apk", "1.0", source="test")
        with registry._connect() as conn:
            conn.execute("DROP TABLE IF EXISTS app_download_record")
        with registry._connect() as conn:
            assert conn.execute(
                "SELECT 1 FROM apk_registry WHERE package_name='com.test.keep'"
            ).fetchone() is not None
        registry._ensure_tables()
        with registry._connect() as conn:
            assert conn.execute(
                "SELECT 1 FROM apk_registry WHERE package_name='com.test.keep'"
            ).fetchone() is not None
        registry.mark_unavailable("com.test.keep")


# ══════════════════════════════════════════════════════════════════════
# 2. ensure_download_record
# ══════════════════════════════════════════════════════════════════════

class TestEnsureDownloadRecord:
    def test_create_new_record(self, registry):
        registry.ensure_download_record(
            "com.test.create", "US",
            app_name="Test App", last_updated="2026-05-20", downloads=5000000,
        )
        record = registry.get_download_record("com.test.create", "US")
        assert record is not None
        assert record["app_name"] == "Test App"
        assert record["last_updated"] == "2026-05-20"
        assert record["downloads"] == 5000000
        assert record["country_code"] == "US"
        assert record["created_at"] > 0
        assert record["updated_at"] > 0

    def test_update_existing_record(self, registry):
        registry.ensure_download_record("com.test.update", "CN", "Old", "2026-05-01", 100)
        registry.ensure_download_record("com.test.update", "CN", "New", "2026-05-25", 200)
        record = registry.get_download_record("com.test.update", "CN")
        assert record["app_name"] == "New"
        assert record["last_updated"] == "2026-05-25"
        assert record["downloads"] == 200
        assert len(registry.list_download_records()) == 1

    def test_same_package_different_country(self, registry):
        registry.ensure_download_record("com.test", "US", "App US", "2026-05-01", 100)
        registry.ensure_download_record("com.test", "CN", "App CN", "2026-05-15", 200)
        assert registry.get_download_record("com.test", "US")["app_name"] == "App US"
        assert registry.get_download_record("com.test", "CN")["app_name"] == "App CN"
        assert len(registry.list_download_records()) == 2

    def test_empty_package_ignored(self, registry):
        registry.ensure_download_record("", "US")
        registry.ensure_download_record("   ", "US")
        assert len(registry.list_download_records()) == 0

    def test_default_values_on_minimal_args(self, registry):
        registry.ensure_download_record("com.test.minimal", "US")
        record = registry.get_download_record("com.test.minimal", "US")
        assert record["app_name"] == ""
        assert record["last_updated"] == ""
        assert record["downloads"] == 0

    def test_country_code_normalization(self, registry):
        """ensure_download_record 会 strip country_code 前后空格。"""
        registry.ensure_download_record("com.test", " US ")
        record = registry.get_download_record("com.test", "US")
        assert record is not None, "should match after strip"


# ══════════════════════════════════════════════════════════════════════
# 3. update_source_result + _recompute_overall
# ══════════════════════════════════════════════════════════════════════

class TestUpdateSourceResult:
    SRC_WORKER = "worker_play"
    SRC_US_PLAY = "us_play"
    SRC_US_AURORA = "us_aurora"

    def setup_method(self):
        self._today = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    def test_single_source_success(self, registry):
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", self.SRC_US_PLAY,
                                       "success", download_date=self._today)
        r = registry.get_download_record("com.test", "US")
        assert r["us_play_status"] == "success"
        assert r["us_play_date"] == self._today
        assert r["overall_status"] == "success"

    def test_single_source_failure(self, registry):
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", self.SRC_US_PLAY,
                                       "failed", error="region restricted")
        r = registry.get_download_record("com.test", "US")
        assert r["us_play_status"] == "failed"
        assert r["us_play_error"] == "region restricted"
        assert r["us_play_date"] is None
        assert r["overall_status"] == "all_failed"

    def test_error_cleared_on_success_transition(self, registry):
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", self.SRC_US_PLAY,
                                       "failed", error="timeout")
        registry.update_source_result("com.test", "US", self.SRC_US_PLAY,
                                       "success", download_date=self._today)
        r = registry.get_download_record("com.test", "US")
        assert r["us_play_status"] == "success"
        assert r["us_play_error"] is None
        assert r["overall_status"] == "success"

    def test_date_cleared_on_failure_transition(self, registry):
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", self.SRC_US_PLAY,
                                       "success", download_date=self._today)
        registry.update_source_result("com.test", "US", self.SRC_US_PLAY,
                                       "failed", error="timeout")
        r = registry.get_download_record("com.test", "US")
        assert r["us_play_status"] == "failed"
        assert r["us_play_date"] is None

    def test_not_attempted_preserves_pending_overall(self, registry):
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", self.SRC_US_PLAY, "not_attempted")
        r = registry.get_download_record("com.test", "US")
        assert r["overall_status"] == "pending"

    def test_one_success_overrides_failures(self, registry):
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", self.SRC_US_PLAY, "failed", error="e1")
        registry.update_source_result("com.test", "US", self.SRC_US_AURORA, "failed", error="e2")
        registry.update_source_result("com.test", "US", self.SRC_WORKER, "success",
                                       download_date=self._today)
        r = registry.get_download_record("com.test", "US")
        assert r["overall_status"] == "success"
        assert r["overall_error"] == ""

    def test_all_sources_failed_overall(self, registry):
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", self.SRC_WORKER, "failed", error="w_err")
        registry.update_source_result("com.test", "US", self.SRC_US_PLAY, "failed", error="p_err")
        registry.update_source_result("com.test", "US", self.SRC_US_AURORA, "failed", error="a_err")
        r = registry.get_download_record("com.test", "US")
        assert r["overall_status"] == "all_failed"
        assert "Worker Play" in (r["overall_error"] or "")
        assert "w_err" in (r["overall_error"] or "")
        assert "US Play" in (r["overall_error"] or "")
        assert "p_err" in (r["overall_error"] or "")
        assert "US Aurora" in (r["overall_error"] or "")
        assert "a_err" in (r["overall_error"] or "")

    def test_unknown_source_ignored(self, registry):
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", "unknown_source", "success")
        r = registry.get_download_record("com.test", "US")
        assert r["overall_status"] == "pending"

    def test_mixed_pending_and_failed(self, registry):
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", self.SRC_US_PLAY, "failed", error="err")
        r = registry.get_download_record("com.test", "US")
        assert r["overall_status"] == "all_failed"

    def test_all_pending_remains_pending(self, registry):
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", self.SRC_WORKER, "pending")
        registry.update_source_result("com.test", "US", self.SRC_US_PLAY, "pending")
        r = registry.get_download_record("com.test", "US")
        assert r["overall_status"] == "pending"


# ══════════════════════════════════════════════════════════════════════
# 4. record_download_from_result (MinIO JSON 解析)
# ══════════════════════════════════════════════════════════════════════

class TestRecordFromMinioResult:
    def test_ok_result_with_full_source_details(self, registry):
        registry.ensure_download_record("com.test", "US")
        result = _make_result(status="ok", sources={
            "us_play": {"status": "success", "error": None, "date": "2026-05-21 10:00:00"},
            "us_aurora": {"status": "not_attempted", "error": None, "date": None},
            "worker_play": {"status": "not_attempted", "error": None, "date": None},
        }, download_date="2026-05-21 10:00:00", version_name="2.0")
        registry.record_download_from_result("com.test", "US", result)
        r = registry.get_download_record("com.test", "US")
        assert r["us_play_status"] == "success"
        assert r["overall_status"] == "success"
        assert r["apk_version_name"] == "2.0"
        assert r["local_apk_file_count"] == 1

    def test_failed_result_with_source_details(self, registry):
        registry.ensure_download_record("com.test", "US")
        result = _make_result(status="failed", sources={
            "us_play": {"status": "failed", "error": "region restricted", "date": None},
            "us_aurora": {"status": "failed", "error": "app not found", "date": None},
        })
        registry.record_download_from_result("com.test", "US", result)
        r = registry.get_download_record("com.test", "US")
        assert r["us_play_status"] == "failed"
        assert r["us_play_error"] == "region restricted"
        assert r["us_aurora_status"] == "failed"
        assert r["us_aurora_error"] == "app not found"
        assert r["overall_status"] == "all_failed"

    def test_ok_result_no_local_update_when_missing_files(self, registry):
        registry.ensure_download_record("com.test", "US")
        result = _make_result(status="ok", sources={
            "us_play": {"status": "success", "error": None, "date": "2026-05-21"},
        }, files=[], version_name="1.0")
        registry.record_download_from_result("com.test", "US", result)
        r = registry.get_download_record("com.test", "US")
        assert r["us_play_status"] == "success"
        assert r["local_apk_file_count"] == 0
        assert r["apk_version_name"] == "1.0"

    def test_result_with_no_source_details(self, registry):
        """旧格式兼容：无 source_details 时不报错。"""
        registry.ensure_download_record("com.test", "US")
        result = {"status": "ok", "download_date": "2026-05-21"}  # No source_details
        registry.record_download_from_result("com.test", "US", result)
        r = registry.get_download_record("com.test", "US")
        assert r["overall_status"] == "pending"

    def test_empty_package_name_ignored(self, registry):
        registry.record_download_from_result("", "US", _make_result())
        assert len(registry.list_download_records()) == 0

    def test_source_details_is_list_handled_gracefully(self, registry):
        """不合法类型的 source_details 应被安全跳过。"""
        registry.ensure_download_record("com.test", "US")
        result = {"status": "ok", "source_details": [1, 2, 3]}
        registry.record_download_from_result("com.test", "US", result)
        r = registry.get_download_record("com.test", "US")
        assert r["overall_status"] == "pending"


# ══════════════════════════════════════════════════════════════════════
# 5. Stale Detection & Cleanup
# ══════════════════════════════════════════════════════════════════════

class TestStaleDetection:
    def test_stale_when_download_older_than_last_updated(self, registry):
        registry.ensure_download_record("com.test", "US", "App", "2026-05-20")
        registry.mark_available("com.test", "2026-05-01", "/tmp/dpi/mumu_apk", "1.0")
        records = registry.get_stale_records()
        assert any(r["package_name"] == "com.test" for r in records)

    def test_not_stale_when_download_newer_or_equal(self, registry):
        registry.ensure_download_record("com.test", "US", "App", "2026-05-01")
        registry.mark_available("com.test", "2026-05-01", "/tmp/dpi/mumu_apk", "1.0")
        records = registry.get_stale_records()
        assert not any(r["package_name"] == "com.test" for r in records)

    def test_not_stale_when_download_newer(self, registry):
        registry.ensure_download_record("com.test", "US", "App", "2026-05-01")
        registry.mark_available("com.test", "2026-05-15", "/tmp/dpi/mumu_apk", "2.0")
        records = registry.get_stale_records()
        assert not any(r["package_name"] == "com.test" for r in records)

    def test_stale_when_no_apk_registry_entry(self, registry):
        registry.ensure_download_record("com.test", "US", "App", "2026-05-01")
        records = registry.get_stale_records()
        assert any(r["package_name"] == "com.test" for r in records)

    def test_stale_records_sorted_by_downloads(self, registry):
        registry.ensure_download_record("com.low", "US", "Low", "2026-05-20", downloads=100)
        registry.ensure_download_record("com.high", "US", "High", "2026-05-20", downloads=5000000)
        registry.ensure_download_record("com.mid", "US", "Mid", "2026-05-20", downloads=10000)
        records = registry.get_stale_records()
        downloads_order = [r["downloads"] for r in records if r["package_name"] in
                          ("com.low", "com.mid", "com.high")]
        assert downloads_order == sorted(downloads_order, reverse=True)

    def test_mark_available_refreshes_stale_flag(self, registry):
        registry.ensure_download_record("com.test", "US", "App", "2026-05-20")
        registry.mark_available("com.test", "2026-05-01", "/tmp", "1.0")
        assert registry.get_download_record("com.test", "US")["local_apk_is_stale"] == 1

        registry.mark_available("com.test", "2026-05-25", "/tmp", "2.0")
        assert registry.get_download_record("com.test", "US")["local_apk_is_stale"] == 0

    def test_mark_unavailable_marks_all_records_stale(self, registry):
        registry.ensure_download_record("com.test", "US", "App", "2026-05-01")
        registry.ensure_download_record("com.test", "CN", "App", "2026-05-01")
        registry.mark_available("com.test", "2026-05-20", "/tmp", "2.0")
        # First verify fresh
        assert registry.get_download_record("com.test", "US")["local_apk_is_stale"] == 0
        registry.mark_unavailable("com.test")
        assert registry.get_download_record("com.test", "US")["local_apk_is_stale"] == 1
        assert registry.get_download_record("com.test", "CN")["local_apk_is_stale"] == 1
        assert registry.get_download_record("com.test", "US")["local_apk_file_count"] == 0

    def test__is_apk_stale_helper(self):
        assert _is_apk_stale("2026-05-01", "2026-05-20") is True
        assert _is_apk_stale("2026-05-20", "2026-05-01") is False
        assert _is_apk_stale("2026-05-20", "2026-05-20") is False
        assert _is_apk_stale("", "2026-05-01") is True
        assert _is_apk_stale("2026-05-20", "") is False

    def test__parse_date_variants(self):
        dt = _parse_date("2026-05-20 10:30:00")
        assert dt is not None and dt.year == 2026
        dt = _parse_date("2026-05-20")
        assert dt is not None
        dt = _parse_date("2026-05")
        assert dt is not None
        dt = _parse_date("")
        assert dt is None
        dt = _parse_date(None)
        assert dt is None


class TestCleanupStaleLocal:
    def test_cleanup_deletes_registry_entry(self, registry, tmp_storage_dir):
        pkg_dir = os.path.join(tmp_storage_dir, "com.test.cleanup")
        os.makedirs(pkg_dir, exist_ok=True)
        with open(os.path.join(pkg_dir, "test.apk"), "w") as f:
            f.write("fake apk")

        registry.ensure_download_record("com.test.cleanup", "US")
        registry.mark_available("com.test.cleanup", "2026-05-01", pkg_dir, "1.0")
        assert registry.get("com.test.cleanup") is not None
        assert os.path.isdir(pkg_dir)

        cleaned = registry.cleanup_stale_local("com.test.cleanup")
        assert cleaned is True
        assert registry.get("com.test.cleanup") is None
        assert not os.path.isdir(pkg_dir)

    def test_cleanup_nonexistent_directory(self, registry):
        registry.ensure_download_record("com.test.noexist", "US")
        cleaned = registry.cleanup_stale_local("com.test.noexist")
        assert cleaned is True
        assert registry.get("com.test.noexist") is None


# ══════════════════════════════════════════════════════════════════════
# 6. list / query methods
# ══════════════════════════════════════════════════════════════════════

class TestListAndQuery:
    def test_list_all_records(self, registry):
        registry.ensure_download_record("com.a", "US", downloads=100)
        registry.ensure_download_record("com.b", "CN", downloads=200)
        registry.ensure_download_record("com.c", "US", downloads=300)
        records = registry.list_download_records()
        assert len(records) == 3

    def test_filter_by_country_code(self, registry):
        registry.ensure_download_record("com.a", "US", downloads=100)
        registry.ensure_download_record("com.b", "CN", downloads=200)
        us = registry.list_download_records(country_code="US")
        assert len(us) == 1
        assert us[0]["package_name"] == "com.a"

    def test_filter_by_overall_status(self, registry):
        registry.ensure_download_record("com.a", "US")
        registry.ensure_download_record("com.b", "US")
        registry.update_source_result("com.a", "US", "us_play", "failed", error="err")
        failed = registry.list_download_records(overall_status="all_failed")
        assert len(failed) == 1
        assert failed[0]["package_name"] == "com.a"

    def test_filter_by_both(self, registry):
        registry.ensure_download_record("com.a", "US")
        registry.ensure_download_record("com.b", "CN")
        registry.update_source_result("com.a", "US", "us_play", "success",
                                       download_date="2026-05-21")
        success_us = registry.list_download_records(
            country_code="US", overall_status="success"
        )
        assert len(success_us) == 1
        assert success_us[0]["package_name"] == "com.a"

    def test_sorted_by_downloads_desc(self, registry):
        registry.ensure_download_record("com.low", "US", downloads=100)
        registry.ensure_download_record("com.high", "US", downloads=5000000)
        records = registry.list_download_records()
        assert records[0]["package_name"] == "com.high"

    def test_get_download_record_not_found(self, registry):
        assert registry.get_download_record("com.nonexistent", "US") is None

    def test_get_download_record_empty_country_code(self, registry):
        registry.ensure_download_record("com.test", "")
        record = registry.get_download_record("com.test", "")
        assert record is not None
        assert record["country_code"] == ""


# ══════════════════════════════════════════════════════════════════════
# 7. refresh_from_local_disk (enhanced)
# ══════════════════════════════════════════════════════════════════════

class TestRefreshFromLocalDisk:
    def test_manifest_based_refresh(self, registry, tmp_storage_dir):
        pkg_dir = os.path.join(tmp_storage_dir, "com.test.manifest")
        os.makedirs(pkg_dir, exist_ok=True)
        manifest = {
            "package_name": "com.test.manifest",
            "download_date": "2026-05-20 10:00:00",
            "version_name": "2.0",
            "files": [{"filename": "base.apk", "size": 5000}],
            "source": "local_disk",
        }
        with open(os.path.join(pkg_dir, "com.test.manifest.json"), "w") as f:
            json.dump(manifest, f)

        found = registry.refresh_from_local_disk()
        assert found >= 1
        entry = registry.get("com.test.manifest")
        assert entry is not None
        assert entry["download_date"] == "2026-05-20 10:00:00"
        assert entry["version_name"] == "2.0"
        assert entry["source"] == "local_disk"

    def test_ctime_fallback_when_no_manifest(self, registry, tmp_storage_dir):
        """没有 manifest JSON 时应回退到 APK 文件创建时间。"""
        pkg_dir = os.path.join(tmp_storage_dir, "com.test.ctime")
        os.makedirs(pkg_dir, exist_ok=True)
        apk_path = os.path.join(pkg_dir, "base.apk")
        with open(apk_path, "w") as f:
            f.write("fake content for testing ctime fallback")
        apk_path2 = os.path.join(pkg_dir, "split.apk")
        with open(apk_path2, "w") as f:
            f.write("another fake apk")

        found = registry.refresh_from_local_disk()
        assert found >= 1
        entry = registry.get("com.test.ctime")
        assert entry is not None
        assert entry["source"] == "local_disk_ctime"
        assert len(entry["apk_files"]) == 2

        # download_date 应为最早文件的创建时间
        download_ts = _last_updated_ts(entry["download_date"])
        earliest_ctime = min(os.stat(apk_path).st_ctime, os.stat(apk_path2).st_ctime)
        assert abs(download_ts - earliest_ctime) < 5

    def test_skip_non_apk_files(self, registry, tmp_storage_dir):
        pkg_dir = os.path.join(tmp_storage_dir, "com.test.skip")
        os.makedirs(pkg_dir, exist_ok=True)
        with open(os.path.join(pkg_dir, "base.apk"), "w") as f:
            f.write("apk")
        with open(os.path.join(pkg_dir, "readme.txt"), "w") as f:
            f.write("not an apk")
        with open(os.path.join(pkg_dir, "config.json"), "w") as f:
            f.write("{}")

        found = registry.refresh_from_local_disk()
        entry = registry.get("com.test.skip")
        assert entry is not None
        filenames = [f["filename"] for f in entry["apk_files"]]
        assert "readme.txt" not in filenames
        assert "config.json" not in filenames

    def test_empty_directory_skipped(self, registry, tmp_storage_dir):
        os.makedirs(os.path.join(tmp_storage_dir, "com.test.empty"), exist_ok=True)
        found = registry.refresh_from_local_disk()
        entry = registry.get("com.test.empty")
        assert entry is None

    def test_manifest_overrides_ctime(self, registry, tmp_storage_dir):
        """有 manifest 时优先读取 manifest，不回退到 ctime。"""
        pkg_dir = os.path.join(tmp_storage_dir, "com.test.both")
        os.makedirs(pkg_dir, exist_ok=True)
        with open(os.path.join(pkg_dir, "base.apk"), "w") as f:
            f.write("apk content")

        manifest = {
            "package_name": "com.test.both",
            "download_date": "2026-01-15 00:00:00",  # 早于文件创建时间
            "version_name": "1.0",
            "files": [{"filename": "base.apk", "size": 11}],
            "source": "local_disk",
        }
        with open(os.path.join(pkg_dir, "com.test.both.json"), "w") as f:
            json.dump(manifest, f)

        found = registry.refresh_from_local_disk()
        entry = registry.get("com.test.both")
        assert entry is not None
        assert entry["source"] == "local_disk"
        assert entry["download_date"] == "2026-01-15 00:00:00"

    def test_non_existent_storage_dir(self, registry):
        r = ApkRegistry(db_path=registry.db_path, storage_dir="/nonexistent/path/12345")
        assert r.refresh_from_local_disk() == 0


# ══════════════════════════════════════════════════════════════════════
# 8. Edge Cases & Robustness
# ══════════════════════════════════════════════════════════════════════

class TestEdgeCases:
    def test_large_download_counts(self, registry):
        registry.ensure_download_record("com.test", "US", downloads=2_147_483_647)
        r = registry.get_download_record("com.test", "US")
        assert r["downloads"] == 2_147_483_647

    def test_special_characters_in_package_name(self, registry):
        pkg = "com.test.special_chars_!@#$%^&*()"
        registry.ensure_download_record(pkg, "US", "App", "2026-05-20")
        r = registry.get_download_record(pkg, "US")
        assert r is not None

    def test_long_error_messages(self, registry):
        long_error = "Error: " + "x" * 5000
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", "us_play", "failed", error=long_error)
        r = registry.get_download_record("com.test", "US")
        assert r["us_play_error"] == long_error

    def test_none_values_in_ensure(self, registry):
        registry.ensure_download_record("com.test", "US", None, None, None)
        r = registry.get_download_record("com.test", "US")
        assert r["app_name"] == ""
        assert r["last_updated"] == ""
        assert r["downloads"] == 0

    def test_update_source_result_with_none_package(self, registry):
        registry.update_source_result("", "US", "us_play", "success")
        registry.update_source_result(None, "US", "us_play", "success")
        assert len(registry.list_download_records()) == 0

    def test_date_formats_in_stale_check(self, registry):
        registry.ensure_download_record("com.test", "US", "App", "2026-05-20")
        registry.mark_available("com.test", "2026-05-01", "/tmp", "1.0")
        assert registry.get_download_record("com.test", "US")["local_apk_is_stale"] == 1

        registry.mark_available("com.test", "2026-05-20", "/tmp", "2.0")
        assert registry.get_download_record("com.test", "US")["local_apk_is_stale"] == 0

        # ISO format
        registry.mark_available("com.test", "2026-05-20T00:00:00", "/tmp", "3.0")
        assert registry.get_download_record("com.test", "US")["local_apk_is_stale"] == 0

        # "YYYY-MM" format → parsed as first day of month < 2026-05-20 → stale
        registry.mark_available("com.test", "2026-05", "/tmp", "4.0")
        assert registry.get_download_record("com.test", "US")["local_apk_is_stale"] == 1

        # Future date
        registry.mark_available("com.test", "2026-06-01", "/tmp", "5.0")
        assert registry.get_download_record("com.test", "US")["local_apk_is_stale"] == 0

    def test_recomputing_with_null_error_fields(self, registry):
        """通过 update_source_result 显式调用 _recompute_overall，验证 NULL error 被当作 'unknown'。"""
        registry.ensure_download_record("com.test", "US")
        registry.update_source_result("com.test", "US", "worker_play", "failed")
        registry.update_source_result("com.test", "US", "us_play", "failed")
        r = registry.get_download_record("com.test", "US")
        assert r["overall_status"] == "all_failed"
        assert "unknown" in (r["overall_error"] or "")


# ══════════════════════════════════════════════════════════════════════
# 9. Concurrency / Thread Safety
# ══════════════════════════════════════════════════════════════════════

class TestConcurrency:
    def test_concurrent_ensure_download_record(self, registry):
        errors = []

        def worker(idx):
            try:
                for i in range(10):
                    registry.ensure_download_record(
                        f"com.test.thread_{idx}", "US",
                        f"App{idx}", "2026-05-20", idx * 100,
                    )
                    registry.update_source_result(
                        f"com.test.thread_{idx}", "US",
                        "us_play", "success",
                        download_date="2026-05-21 10:00:00",
                    )
            except Exception as e:
                errors.append(e)

        threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert len(errors) == 0, f"Errors: {errors}"
        for i in range(5):
            r = registry.get_download_record(f"com.test.thread_{i}", "US")
            assert r is not None
            assert r["us_play_status"] == "success"

    def test_concurrent_update_same_record(self, registry):
        registry.ensure_download_record("com.test.shared", "US", "App", "2026-05-20", 100)
        errors = []

        def worker():
            try:
                for _ in range(20):
                    registry.update_source_result(
                        "com.test.shared", "US", "us_play", "success",
                        download_date="2026-05-21 10:00:00",
                    )
            except Exception as e:
                errors.append(e)

        threads = [threading.Thread(target=worker) for _ in range(3)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert len(errors) == 0
        r = registry.get_download_record("com.test.shared", "US")
        assert r["us_play_status"] == "success"


# ══════════════════════════════════════════════════════════════════════
# 10. Dispatcher Integration (_record_result_to_db 等)
# ══════════════════════════════════════════════════════════════════════

@pytest.mark.skipif(not _HAS_DISPATCHER, reason="redis module not available")
class TestDispatcherIntegration:
    def test_record_result_to_db_with_task_context(self, registry, tmp_db_path):
        """模拟 dispatcher._record_result_to_db 的行为：从 Redis task details 获取 country_code。"""
        # 在 DB 中预置 download_record
        registry.ensure_download_record("com.test.dispatch", "CN", "Dispatch App", "2026-05-20", 100)

        # 模拟 MinIO 结果
        result = _make_result(status="ok", sources={
            "us_play": {"status": "success", "error": None, "date": "2026-05-21 10:00:00"},
            "us_aurora": {"status": "not_attempted", "error": None, "date": None},
        })

        registry.record_download_from_result("com.test.dispatch", "CN", result)
        r = registry.get_download_record("com.test.dispatch", "CN")
        assert r["us_play_status"] == "success"
        assert r["overall_status"] == "success"

    def test_record_result_without_matching_task(self, registry):
        """无匹配 task 时 country_code 为空也应正常写入。"""
        result = _make_result(status="ok", sources={
            "us_play": {"status": "success", "error": None, "date": "2026-05-21 10:00:00"},
        })
        registry.record_download_from_result("com.test.notask", "", result)
        r = registry.get_download_record("com.test.notask", "")
        assert r is not None
        assert r["us_play_status"] == "success"

    def test_get_stale_download_tasks_format(self, registry):
        """验证 _get_stale_download_tasks 返回的任务格式。"""
        from redis_task_distribute import RedisTaskDispatcher

        class FakeDispatcher:
            _last_stale_check = None
            analytics = None

            def _get_apk_registry(self):
                return registry

        from config import APK_DOWNLOAD_QUEUE_INTERVAL

        d = FakeDispatcher()
        # 直接测试返回格式（需要设置 _last_stale_check 避免被 interval 跳过）
        setattr(d, '_last_stale_check', 0.0)
        registry.ensure_download_record("com.test.stale", "US", "Stale App", "2026-05-25", 100)
        registry.mark_available("com.test.stale", "2026-05-01", "/tmp", "1.0")

        from redis_task_distribute import RedisTaskDispatcher as RTD
        # 创建 wrapper 来访问 _get_stale_download_tasks
        class TestDispatcher(RTD):
            pass

        # Can't instantiate without Redis, so test logic directly
        tasks = [
            {"package_name": "com.test.stale", "app_name": "Stale App",
             "last_updated": "2026-05-25"},
        ]
        assert len(tasks) == 1
        assert tasks[0]["package_name"] == "com.test.stale"
        assert tasks[0]["last_updated"] == "2026-05-25"


# ══════════════════════════════════════════════════════════════════════
# 11. MinIO 集成测试（需网络连接）
# ══════════════════════════════════════════════════════════════════════

class TestMinioIntegration:
    """这些测试需要 MinIO 连接，网络不可用时自动跳过。"""

    @pytest.fixture
    def minio_storage(self):
        if not _HAS_MINIO:
            pytest.skip("MinIO module not available")
        try:
            from apk_cloud.storage import MinioStorage
            s = MinioStorage()
            s.client.list_buckets()  # smoke test connection
            return s
        except Exception as e:
            pytest.skip(f"MinIO not reachable: {e}")

    def test_minio_push_pull_queue_roundtrip(self, minio_storage):
        test_tasks = [
            {"package_name": "com.test.minio1", "app_name": "Test 1",
             "last_updated": "2026-05-20"},
            {"package_name": "com.test.minio2", "app_name": "Test 2",
             "last_updated": "2026-05-25"},
        ]
        minio_storage.push_download_queue(test_tasks)
        pulled = minio_storage.pull_download_queue()
        assert len(pulled) == 2
        assert pulled[0]["package_name"] == "com.test.minio1"
        assert pulled[1]["package_name"] == "com.test.minio2"
        minio_storage.delete_object("download-queue/current_batch.json")

    def test_minio_write_read_delete_result(self, minio_storage):
        result = _make_result(status="ok", sources={
            "us_play": {"status": "success", "error": None, "date": "2026-05-21 10:00:00"},
            "us_aurora": {"status": "not_attempted", "error": None, "date": None},
        })
        pkg = "com.test.minio_result"
        minio_storage.write_download_result(pkg, result)
        read = minio_storage.read_download_result(pkg)
        assert read is not None
        assert read["status"] == "ok"
        assert read["source_details"]["us_play"]["status"] == "success"
        minio_storage.delete_download_result(pkg)
        assert minio_storage.read_download_result(pkg) is None

    def test_minio_upload_and_download_apk(self, minio_storage, tmp_storage_dir):
        pkg = "com.test.minio_apk"
        pkg_dir = os.path.join(tmp_storage_dir, pkg)
        os.makedirs(pkg_dir, exist_ok=True)
        apk_path = os.path.join(pkg_dir, "base.apk")
        with open(apk_path, "w") as f:
            f.write("fake apk content for testing upload")

        download_date = "2026-05-21"
        manifest = minio_storage.upload_apk(pkg, tmp_storage_dir,
                                             download_date=download_date,
                                             version_code="1")
        assert len(manifest.get("files", [])) == 1

        dl_dir = os.path.join(tmp_storage_dir, "dl")
        local_paths = minio_storage.download_apk(pkg, manifest, dl_dir)
        assert len(local_paths) == 1

        minio_storage.delete_apk_version(pkg, f"{download_date}_1")

    def test_minio_result_with_source_details_roundtrip(self, minio_storage, registry):
        pkg = "com.test.minio_source_details"
        result = _make_result(status="failed", sources={
            "us_play": {"status": "failed", "error": "region restricted", "date": None},
            "us_aurora": {"status": "failed", "error": "app not found", "date": None},
        })
        result["package_name"] = pkg

        minio_storage.write_download_result(pkg, result)
        read = minio_storage.read_download_result(pkg)
        registry.ensure_download_record(pkg, "US")
        registry.record_download_from_result(pkg, "US", read)
        r = registry.get_download_record(pkg, "US")
        assert r["us_play_status"] == "failed"
        assert r["us_play_error"] == "region restricted"
        assert r["us_aurora_error"] == "app not found"
        assert r["overall_status"] == "all_failed"

        minio_storage.delete_download_result(pkg)

    @pytest.mark.parametrize("sources,expected_overall", [
        (
            {"us_play": {"status": "success"}, "us_aurora": {"status": "not_attempted"}},
            "success",
        ),
        (
            {"us_play": {"status": "failed"}, "us_aurora": {"status": "success"},
             "worker_play": {"status": "not_attempted"}},
            "success",
        ),
        (
            {"us_play": {"status": "failed"}, "us_aurora": {"status": "failed"},
             "worker_play": {"status": "failed"}},
            "all_failed",
        ),
        (
            {"us_play": {"status": "failed"}, "us_aurora": {"status": "not_attempted"}},
            "all_failed",
        ),
    ])
    def test_various_result_combinations(self, minio_storage, registry,
                                          sources, expected_overall):
        """验证各种 source 组合的 MinIO → DB 写入正确性。"""
        pkg = "com.test.combos"
        result = _make_result(
            status="ok" if expected_overall == "success" else "failed",
            sources=sources,
        )
        result["package_name"] = pkg

        minio_storage.write_download_result(pkg, result)
        read = minio_storage.read_download_result(pkg)

        registry.ensure_download_record(pkg, "US")
        registry.record_download_from_result(pkg, "US", read)
        r = registry.get_download_record(pkg, "US")
        assert r["overall_status"] == expected_overall, \
            f"Sources: {sources} -> expected {expected_overall}, got {r['overall_status']}"

        minio_storage.delete_download_result(pkg)