"""UserStore / bootstrap_admin 单元测试（FakeRedis 与内存降级，不连真实 Redis）

覆盖：
- 哈希：同密码同 salt 一致；不同 salt 不同
- create：重名 UserExistsError；非法用户名/弱密码 ValueError；记录不含明文密码
- get/list/delete/verify_password/set_password/count_admins
- bootstrap_admin：空库创建 admin 并打印/返回密码；非空库不重复创建
- 内存降级（构造传 None）全功能可用
- Redis 异常包装为 UserStoreError
"""

import fnmatch
import json
from datetime import datetime

import pytest

from app.core.users import (
    UserExistsError,
    UserNotFoundError,
    UserStore,
    UserStoreError,
    bootstrap_admin,
)


class FakeRedis:
    """最小内存版 redis.asyncio 客户端：get/set/setex/delete/scan_iter/keys + TTL 记录

    fail=True 时所有操作抛 ConnectionError，用于验证异常包装语义。
    """

    def __init__(self, fail: bool = False) -> None:
        self.fail = fail
        self.store: dict[str, str] = {}
        self.ttls: dict[str, int] = {}

    def _check(self) -> None:
        if self.fail:
            raise ConnectionError("redis down")

    async def get(self, key: str) -> str | None:
        self._check()
        return self.store.get(key)

    async def set(self, key: str, value: str) -> bool:
        self._check()
        self.store[key] = value
        return True

    async def setex(self, key: str, ttl: int, value: str) -> bool:
        self._check()
        self.store[key] = value
        self.ttls[key] = ttl
        return True

    async def delete(self, *keys: str) -> int:
        self._check()
        deleted = 0
        for key in keys:
            if key in self.store:
                del self.store[key]
                deleted += 1
        return deleted

    async def scan_iter(self, match: str = "*"):
        self._check()
        for key in list(self.store):
            if fnmatch.fnmatch(key, match):
                yield key

    async def keys(self, pattern: str = "*") -> list[str]:
        self._check()
        return [key for key in self.store if fnmatch.fnmatch(key, pattern)]


class FakeLogger:
    """捕获 warning 调用的假 logger（structlog 风格：event + kwargs）"""

    def __init__(self) -> None:
        self.warnings: list[tuple[str, dict]] = []

    def warning(self, event: str, **kwargs) -> None:
        self.warnings.append((event, kwargs))


class TestHashPassword:
    """PBKDF2 哈希确定性"""

    def test_same_password_same_salt_consistent(self):
        store = UserStore(None)
        salt = b"0123456789abcdef"
        assert store.hash_password("password123", salt) == store.hash_password("password123", salt)

    def test_different_salt_different_hash(self):
        store = UserStore(None)
        assert store.hash_password("password123", b"a" * 16) != store.hash_password("password123", b"b" * 16)


class TestCreate:
    """创建用户：校验、重名、无明文"""

    async def test_create_success(self):
        store = UserStore(FakeRedis())
        record = await store.create("alice", "password123")
        assert record.username == "alice"
        assert record.role == "user"
        assert record.must_change_password is False
        # created_at 为可解析的 ISO8601
        datetime.fromisoformat(record.created_at)

    async def test_record_contains_no_plaintext(self):
        redis = FakeRedis()
        store = UserStore(redis)
        record = await store.create("alice", "password123")
        assert record.password_hash != "password123"
        assert "password123" not in record.password_hash
        # salt 为 16 字节 hex
        assert len(bytes.fromhex(record.salt)) == 16
        # 落库 JSON 同样不含明文密码
        raw = redis.store["user:alice"]
        assert "password123" not in raw
        assert json.loads(raw)["username"] == "alice"

    async def test_duplicate_raises_user_exists(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "password123")
        with pytest.raises(UserExistsError):
            await store.create("alice", "another-password")

    @pytest.mark.parametrize("username", ["a", "x" * 33, "bad name", "bad!name", "中文名", ""])
    async def test_invalid_username_raises_value_error(self, username: str):
        with pytest.raises(ValueError):
            await UserStore(FakeRedis()).create(username, "password123")

    @pytest.mark.parametrize("username", ["ab", "a" * 32, "A-Z_0-9"])
    async def test_valid_username_boundary(self, username: str):
        record = await UserStore(FakeRedis()).create(username, "password123")
        assert record.username == username

    @pytest.mark.parametrize("password", ["", "short", "1234567"])
    async def test_short_password_raises_value_error(self, password: str):
        with pytest.raises(ValueError):
            await UserStore(FakeRedis()).create("alice", password)

    async def test_exact_min_password_length_ok(self):
        record = await UserStore(FakeRedis()).create("alice", "12345678")
        assert record.username == "alice"


class TestGetListDelete:
    """查询 / 列表 / 删除"""

    async def test_get_roundtrip_and_miss(self):
        store = UserStore(FakeRedis())
        created = await store.create("alice", "password123")
        assert await store.get("alice") == created
        assert await store.get("nobody") is None

    async def test_list_scans_only_user_keys(self):
        redis = FakeRedis()
        store = UserStore(redis)
        await store.create("alice", "password123")
        await store.create("bob", "password456", role="admin")
        redis.store["other:key"] = "not-a-user"  # 非 user: 前缀不参与
        names = {r.username for r in await store.list()}
        assert names == {"alice", "bob"}

    async def test_delete_idempotent(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "password123")
        assert await store.delete("alice") is True
        assert await store.get("alice") is None
        assert await store.delete("alice") is False


class TestVerifyPassword:
    """密码校验"""

    async def test_success_returns_record(self):
        store = UserStore(FakeRedis())
        created = await store.create("alice", "password123")
        assert await store.verify_password("alice", "password123") == created

    async def test_wrong_password_returns_none(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "password123")
        assert await store.verify_password("alice", "wrong-password") is None

    async def test_unknown_user_returns_none(self):
        assert await UserStore(FakeRedis()).verify_password("nobody", "password123") is None


class TestSetPassword:
    """重置密码"""

    async def test_reset_clears_flag_and_invalidates_old(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "old-password", must_change_password=True)
        assert await store.set_password("alice", "new-password") is True
        record = await store.get("alice")
        assert record is not None
        assert record.must_change_password is False
        assert await store.verify_password("alice", "old-password") is None
        assert await store.verify_password("alice", "new-password") is not None

    async def test_unknown_user_returns_false(self):
        assert await UserStore(FakeRedis()).set_password("nobody", "password123") is False

    async def test_short_password_raises_value_error(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "password123")
        with pytest.raises(ValueError):
            await store.set_password("alice", "short")


class TestCountAdmins:
    async def test_counts_only_admin_role(self):
        store = UserStore(FakeRedis())
        assert await store.count_admins() == 0
        await store.create("admin1", "password123", role="admin")
        await store.create("admin2", "password123", role="admin")
        await store.create("user1", "password123")
        assert await store.count_admins() == 2
        await store.delete("admin1")
        assert await store.count_admins() == 1


class TestEnabledField:
    """enabled 字段：默认 True、create 传参、get/list 回读、存量兼容"""

    async def test_default_enabled_is_true(self):
        record = await UserStore(FakeRedis()).create("alice", "password123")
        assert record.enabled is True

    async def test_create_with_enabled_false(self):
        redis = FakeRedis()
        store = UserStore(redis)
        record = await store.create("alice", "password123", enabled=False)
        assert record.enabled is False
        # 持久化后回读仍为 False
        assert (await store.get("alice")).enabled is False
        # 落库 JSON 含 enabled 字段
        assert json.loads(redis.store["user:alice"])["enabled"] is False

    async def test_get_list_roundtrip_enabled(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "password123", enabled=False)
        await store.create("bob", "password123", role="admin")
        assert (await store.get("alice")).enabled is False
        assert (await store.get("bob")).enabled is True
        records = {r.username: r for r in await store.list()}
        assert records["alice"].enabled is False
        assert records["bob"].enabled is True

    async def test_legacy_record_without_enabled_defaults_true(self):
        """存量记录无 enabled 字段时按 True 兼容（get/list）"""
        redis = FakeRedis()
        store = UserStore(redis)
        # 直接写入无 enabled 字段的存量记录
        redis.store["user:legacy"] = json.dumps(
            {
                "username": "legacy",
                "role": "user",
                "password_hash": "hash",
                "salt": "00" * 16,
                "must_change_password": False,
                "created_at": "2024-01-01T00:00:00+00:00",
            }
        )
        record = await store.get("legacy")
        assert record is not None
        assert record.enabled is True
        records = await store.list()
        assert records[0].enabled is True


class TestUpdateUser:
    """update_user：角色/启用状态更新与校验"""

    async def test_update_role(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "password123")
        updated = await store.update_user("alice", role="admin")
        assert updated.role == "admin"
        assert updated.enabled is True  # 未改动
        # 持久化
        record = await store.get("alice")
        assert record is not None
        assert record.role == "admin"

    async def test_update_enabled(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "password123", role="admin")
        updated = await store.update_user("alice", enabled=False)
        assert updated.enabled is False
        assert updated.role == "admin"  # 未改动

    async def test_update_both(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "password123")
        updated = await store.update_user("alice", role="admin", enabled=False)
        assert updated.role == "admin"
        assert updated.enabled is False

    async def test_user_not_found_raises(self):
        with pytest.raises(UserNotFoundError):
            await UserStore(FakeRedis()).update_user("nobody", role="admin")

    async def test_invalid_role_raises_value_error(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "password123")
        with pytest.raises(ValueError):
            await store.update_user("alice", role="superuser")

    async def test_no_fields_noop(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "password123", role="admin")
        updated = await store.update_user("alice")
        assert updated.role == "admin"
        assert updated.enabled is True


class TestBootstrapAdmin:
    """空库引导创建默认管理员"""

    async def test_empty_store_creates_admin_and_returns_password(self):
        store = UserStore(FakeRedis())
        logger = FakeLogger()
        password = await bootstrap_admin(store, logger)  # type: ignore[arg-type]
        assert isinstance(password, str) and len(password) > 0
        record = await store.get("admin")
        assert record is not None
        assert record.role == "admin"
        assert record.must_change_password is True
        # 返回的明文密码可直接通过校验
        assert await store.verify_password("admin", password) is not None
        # 明文密码仅在日志中打印一次
        assert len(logger.warnings) == 1
        assert logger.warnings[0][1].get("password") == password

    async def test_non_empty_store_returns_none_without_changes(self):
        store = UserStore(FakeRedis())
        await store.create("alice", "password123")
        logger = FakeLogger()
        assert await bootstrap_admin(store, logger) is None  # type: ignore[arg-type]
        assert [r.username for r in await store.list()] == ["alice"]
        assert logger.warnings == []

    async def test_second_call_returns_none(self):
        store = UserStore(FakeRedis())
        logger = FakeLogger()
        assert await bootstrap_admin(store, logger) is not None  # type: ignore[arg-type]
        assert await bootstrap_admin(store, logger) is None  # type: ignore[arg-type]
        assert len(await store.list()) == 1


class TestMemoryFallback:
    """构造传 None：纯内存 dict 降级，全功能可用"""

    async def test_full_flow_without_redis(self):
        store = UserStore(None)
        await store.create("alice", "password123", role="admin")
        await store.create("bob", "password456")
        assert await store.count_admins() == 1
        assert {r.username for r in await store.list()} == {"alice", "bob"}
        assert await store.verify_password("alice", "password123") is not None
        assert await store.verify_password("alice", "wrong-password") is None
        assert await store.set_password("bob", "new-password") is True
        assert await store.verify_password("bob", "new-password") is not None
        assert await store.delete("bob") is True
        assert await store.get("bob") is None

    async def test_bootstrap_admin_memory(self):
        store = UserStore(None)
        password = await bootstrap_admin(store, FakeLogger())  # type: ignore[arg-type]
        assert password is not None
        record = await store.get("admin")
        assert record is not None and record.role == "admin"


class TestRedisErrors:
    """Redis 读写异常统一包装为 UserStoreError（不静默）"""

    async def test_create_raises_store_error(self):
        with pytest.raises(UserStoreError):
            await UserStore(FakeRedis(fail=True)).create("alice", "password123")

    async def test_get_raises_store_error(self):
        with pytest.raises(UserStoreError):
            await UserStore(FakeRedis(fail=True)).get("alice")

    async def test_list_raises_store_error(self):
        with pytest.raises(UserStoreError):
            await UserStore(FakeRedis(fail=True)).list()

    async def test_delete_raises_store_error(self):
        with pytest.raises(UserStoreError):
            await UserStore(FakeRedis(fail=True)).delete("alice")

    async def test_set_password_raises_store_error(self):
        with pytest.raises(UserStoreError):
            await UserStore(FakeRedis(fail=True)).set_password("alice", "password123")

    async def test_count_admins_raises_store_error(self):
        with pytest.raises(UserStoreError):
            await UserStore(FakeRedis(fail=True)).count_admins()