import base64
import inspect
from datetime import datetime, timedelta, timezone

import pytest


def test_captcha_missing_wrong_and_second_use():
    from app.services.captcha import CAPTCHA_ALPHABET, consume_captcha, issue_captcha
    from app.services.redis_client import get_redis

    assert consume_captcha("", "ab12") is False
    assert consume_captcha("missing-id", "") is False
    assert consume_captcha("", "") is False

    try:
        client = get_redis()
        client.ping()
    except Exception:
        pytest.skip("Redis 不可达")

    wrong = issue_captcha()
    wrong_id = wrong["captchaId"]
    assert wrong["image"].startswith("data:image/png;base64,")
    raw = base64.b64decode(wrong["image"].split(",", 1)[1])
    assert raw.startswith(b"\x89PNG")
    stored = client.get(f"gsfdc:captcha:{wrong_id}")
    assert isinstance(stored, str)
    assert len(stored) == 4
    assert set(stored).issubset(set(CAPTCHA_ALPHABET.lower()))
    ttl = client.ttl(f"gsfdc:captcha:{wrong_id}")
    assert 0 < ttl <= 120
    assert consume_captcha(wrong_id, "0000") is False
    assert client.get(f"gsfdc:captcha:{wrong_id}") is None
    assert consume_captcha(wrong_id, stored) is False

    ok = issue_captcha()
    ok_id = ok["captchaId"]
    stored_ok = client.get(f"gsfdc:captcha:{ok_id}")
    assert consume_captcha(ok_id, stored_ok.upper()) is True
    assert consume_captcha(ok_id, stored_ok) is False


def test_is_currently_locked_respects_utc_and_does_not_mutate():
    from types import SimpleNamespace

    from app.services.login_guard import is_currently_locked

    user = SimpleNamespace(failed_login_count=2, locked_until=None)
    assert is_currently_locked(user) is False
    assert user.failed_login_count == 2

    user.locked_until = datetime.now(timezone.utc) + timedelta(minutes=1)
    assert is_currently_locked(user) is True
    assert user.failed_login_count == 2

    user.locked_until = datetime.now(timezone.utc) - timedelta(seconds=1)
    assert is_currently_locked(user) is False
    user.locked_until = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(seconds=1)
    assert is_currently_locked(user) is False
    assert user.failed_login_count == 2


def test_lock_after_small_limit_and_past_deadline():
    pytest.importorskip("app.db")
    pytest.importorskip("app.models.content")
    pytest.importorskip("app.models.rbac")

    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

    from app.models.content import SiteSetting
    from app.models.rbac import AdminUser
    from app.services.login_guard import is_currently_locked, read_login_policy, record_failure

    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    SiteSetting.__table__.create(engine)
    AdminUser.__table__.create(engine)
    db = sessionmaker(bind=engine, expire_on_commit=False)()
    try:
        assert read_login_policy(db) == (5, 15)
        db.add(SiteSetting(key="login_max_failures", value="2"))
        db.add(SiteSetting(key="login_lock_minutes", value="3"))
        db.commit()

        user = _new_user(AdminUser, "sec-lock-test")
        db.add(user)
        db.commit()
        db.refresh(user)

        assert is_currently_locked(user) is False
        record_failure(db, user)
        assert user.failed_login_count == 1
        assert user.locked_until is None
        assert is_currently_locked(user) is False
        assert user.failed_login_count == 1

        record_failure(db, user)
        assert user.failed_login_count == 0
        assert user.locked_until is not None
        locked_until = user.locked_until
        if locked_until.tzinfo is not None:
            locked_until = locked_until.astimezone(timezone.utc).replace(tzinfo=None)
        delta = locked_until - datetime.now(timezone.utc).replace(tzinfo=None)
        assert timedelta(minutes=2) < delta < timedelta(minutes=4)
        assert is_currently_locked(user) is True
        assert user.failed_login_count == 0

        user.locked_until = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(seconds=30)
        db.add(user)
        db.commit()
        db.refresh(user)
        assert is_currently_locked(user) is False
        assert user.failed_login_count == 0
    finally:
        db.close()


def test_write_audit_does_not_accept_secret_fields():
    from app.services.audit import write_audit

    signature = inspect.signature(write_audit)
    names = list(signature.parameters)
    assert names == ["user_id", "username", "method", "path", "status_code", "ip"]
    for parameter in signature.parameters.values():
        assert parameter.kind is inspect.Parameter.KEYWORD_ONLY
    banned = {"body", "password", "captcha", "captcha_id", "id_card", "mobile", "phone"}
    assert banned.isdisjoint(names)
    with pytest.raises(TypeError):
        write_audit(
            user_id=None,
            username=None,
            method="POST",
            path="/api/admin/auth/login",
            status_code=400,
            ip="127.0.0.1",
            body="x",
        )
    with pytest.raises(TypeError):
        write_audit(
            user_id=None,
            username=None,
            method="POST",
            path="/api/admin/auth/login",
            status_code=400,
            ip="127.0.0.1",
            password="x",
        )
    with pytest.raises(TypeError):
        write_audit(
            user_id=None,
            username=None,
            method="POST",
            path="/api/admin/auth/login",
            status_code=400,
            ip="127.0.0.1",
            captcha="x",
        )


def test_ip_blacklist_blocks_then_unblocks():
    from app.core.config import get_settings

    if not (get_settings().database_url or "").strip():
        pytest.skip("DATABASE_URL 为空")

    from sqlalchemy import delete

    from app.db import SessionLocal
    from app.models.security import AdminIpBlacklist
    from app.services.ip_guard import is_ip_blocked

    test_ip = "203.0.113.214"
    db = SessionLocal()
    try:
        AdminIpBlacklist.__table__.create(db.get_bind(), checkfirst=True)
        db.execute(delete(AdminIpBlacklist).where(AdminIpBlacklist.ip == test_ip))
        db.commit()
        assert is_ip_blocked(test_ip) is False
        db.add(AdminIpBlacklist(ip=test_ip, remark="security-test"))
        db.commit()
        assert is_ip_blocked(test_ip) is True
        db.execute(delete(AdminIpBlacklist).where(AdminIpBlacklist.ip == test_ip))
        db.commit()
        assert is_ip_blocked(test_ip) is False
    finally:
        try:
            db.rollback()
            db.execute(delete(AdminIpBlacklist).where(AdminIpBlacklist.ip == test_ip))
            db.commit()
        except Exception:
            db.rollback()
        finally:
            db.close()


def _new_user(model, username: str):
    user = model()
    if hasattr(user, "username"):
        user.username = username
    if hasattr(user, "status"):
        user.status = "active"
    if hasattr(user, "failed_login_count"):
        user.failed_login_count = 0
    if hasattr(user, "locked_until"):
        user.locked_until = None
    if hasattr(user, "display_name"):
        user.display_name = "lock-test"
    if hasattr(user, "password_hash"):
        user.password_hash = "test-only-hash"
    for column in user.__table__.columns:
        if column.primary_key or column.nullable:
            continue
        if column.default is not None or column.server_default is not None:
            continue
        if getattr(user, column.key) is not None:
            continue
        try:
            kind = column.type.python_type
        except Exception:
            kind = str
        setattr(user, column.key, 0 if kind is int else "test")
    return user