TelegramBotAdmin/tests/test_giveaway_recurring_eligibility.py

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from types import SimpleNamespace

import pytest
from aiohttp import CookieJar
from aiohttp.test_utils import TestClient, TestServer
from pyrogram.enums import ChatMemberStatus


def _member(status, *, is_member=True):
    return SimpleNamespace(status=status, is_member=is_member, privileges=None)


async def test_eligibility_all_any_and_unknown(app_modules):
    eligibility = app_modules.load("wbb.services.giveaway_eligibility")
    giveaway = {
        "chat_id": -100,
        "eligibility_targets": [{"chat_id": -200}, {"chat_id": -300}],
        "eligibility_mode": "all",
    }
    app_modules.app.members[(-200, 999)] = _member(ChatMemberStatus.ADMINISTRATOR)
    app_modules.app.members[(-300, 999)] = _member(ChatMemberStatus.ADMINISTRATOR)
    app_modules.app.members[(-200, 10)] = _member(ChatMemberStatus.MEMBER)
    app_modules.app.members[(-300, 10)] = _member(ChatMemberStatus.LEFT)
    assert await eligibility.check_eligibility(giveaway, 10) == (False, [-300])

    giveaway["eligibility_mode"] = "any"
    assert await eligibility.check_eligibility(giveaway, 10) == (True, [])
    app_modules.app.members[(-200, 10)] = _member(ChatMemberStatus.RESTRICTED, is_member=False)
    assert await eligibility.check_eligibility(giveaway, 10) == (False, [-200, -300])

    app_modules.app.members[(-300, 999)] = _member(ChatMemberStatus.MEMBER)
    with pytest.raises(eligibility.EligibilityUnavailable):
        await eligibility.check_eligibility(giveaway, 10)


async def test_paid_tickets_idempotent_cap_weight_and_cancel_refund(app_modules):
    points = app_modules.load("wbb.utils.dbpoints")
    database = app_modules.load("wbb.utils.dbgiveaway")
    service = app_modules.load("wbb.services.giveaways")
    await points.adjust_points(
        chat_id=-100, user_id=10, delta=100, source=points.SOURCE_ADMIN,
        idempotency_key="seed:10", reason="seed",
    )
    giveaway = await database.create_giveaway(
        chat_id=-100, creator_id=1, creator_name="Admin", title="Tickets",
        description="", prizes=[{"name": "Winner", "count": 1}],
        ends_at=datetime.now(UTC) + timedelta(hours=1),
        entry_cost=10, max_tickets_per_user=5,
    )
    giveaway_id = giveaway["giveaway_id"]
    status, _ = await service.join_giveaway(
        giveaway_id, user_id=10, username=None, first_name="Alice"
    )
    assert status == "ok"
    status, _ = await service.buy_additional_tickets(
        giveaway_id, user_id=10, quantity=4, request_id="callback-1"
    )
    assert status == "ok"
    replay, _ = await service.buy_additional_tickets(
        giveaway_id, user_id=10, quantity=4, request_id="callback-1"
    )
    capped, _ = await service.buy_additional_tickets(
        giveaway_id, user_id=10, quantity=1, request_id="callback-2"
    )
    assert (replay, capped) == ("duplicate", "ticket_limit")
    participant = await database.get_participant(giveaway_id, 10)
    assert participant["paid_ticket_count"] == 5
    assert (await points.get_point_account(-100, 10))["balance"] == 50
    assert len(service.pick_winners([participant], giveaway["prizes"])) == 1
    canceled, _, _ = await service.cancel_and_refund_giveaway(
        giveaway_id, chat_id=-100, publish=False
    )
    assert canceled is True
    assert (await points.get_point_account(-100, 10))["balance"] == 100


async def test_first_paid_purchase_can_buy_five_tickets_atomically(app_modules):
    points = app_modules.load("wbb.utils.dbpoints")
    database = app_modules.load("wbb.utils.dbgiveaway")
    service = app_modules.load("wbb.services.giveaways")
    await points.adjust_points(
        chat_id=-100, user_id=10, delta=100, source=points.SOURCE_ADMIN,
        idempotency_key="seed:first-five", reason="seed",
    )
    giveaway = await database.create_giveaway(
        chat_id=-100, creator_id=1, creator_name="Admin", title="Five",
        description="", prizes=[{"name": "Winner", "count": 1}],
        ends_at=datetime.now(UTC) + timedelta(hours=1), entry_cost=10,
    )
    giveaway_id = giveaway["giveaway_id"]
    status, _ = await service.buy_additional_tickets(
        giveaway_id, user_id=10, quantity=5, request_id="first-purchase"
    )
    replay, _ = await service.buy_additional_tickets(
        giveaway_id, user_id=10, quantity=5, request_id="first-purchase"
    )
    assert (status, replay) == ("ok", "duplicate")
    assert await database.count_participants(giveaway_id) == 1
    participant = await database.get_participant(giveaway_id, 10)
    assert participant["paid_ticket_count"] == 5
    assert participant["pending"] is False
    assert (await points.get_point_account(-100, 10))["balance"] == 50


async def test_ticket_purchase_recovers_after_debit_before_ticket_write(app_modules, monkeypatch):
    points = app_modules.load("wbb.utils.dbpoints")
    database = app_modules.load("wbb.utils.dbgiveaway")
    service = app_modules.load("wbb.services.giveaways")
    await points.adjust_points(
        chat_id=-100, user_id=10, delta=100, source=points.SOURCE_ADMIN,
        idempotency_key="seed:recover-ticket", reason="seed",
    )
    giveaway = await database.create_giveaway(
        chat_id=-100, creator_id=1, creator_name="Admin", title="Recover",
        description="", prizes=[{"name": "Winner", "count": 1}],
        ends_at=datetime.now(UTC) + timedelta(hours=1), entry_cost=10,
    )
    original_adjust = service.adjust_points
    interrupted = False

    async def interrupt_after_debit(**kwargs):
        nonlocal interrupted
        result = await original_adjust(**kwargs)
        if kwargs["delta"] < 0 and not interrupted:
            interrupted = True
            raise RuntimeError("simulated restart after debit")
        return result

    monkeypatch.setattr(service, "adjust_points", interrupt_after_debit)
    with pytest.raises(RuntimeError, match="simulated restart"):
        await service.buy_additional_tickets(
            giveaway["giveaway_id"], user_id=10, quantity=5,
            request_id="recover-ticket",
        )
    assert (await points.get_point_account(-100, 10))["balance"] == 50
    monkeypatch.setattr(service, "adjust_points", original_adjust)
    await service.reconcile_pending_ticket_orders()
    participant = await database.get_participant(giveaway["giveaway_id"], 10)
    assert participant["paid_ticket_count"] == 5
    assert participant["pending"] is False
    assert (await points.get_point_account(-100, 10))["balance"] == 50


async def test_draw_waits_for_membership_recovery_and_excludes_confirmed_leaver(app_modules):
    points = app_modules.load("wbb.utils.dbpoints")
    database = app_modules.load("wbb.utils.dbgiveaway")
    service = app_modules.load("wbb.services.giveaways")
    for user_id in (10, 11):
        await points.adjust_points(
            chat_id=-100, user_id=user_id, delta=20, source=points.SOURCE_ADMIN,
            idempotency_key=f"seed:eligibility:{user_id}", reason="seed",
        )
    giveaway = await database.create_giveaway(
        chat_id=-100, creator_id=1, creator_name="Admin", title="Membership",
        description="", prizes=[{"name": "Winner", "count": 1}],
        ends_at=datetime.now(UTC) + timedelta(hours=1), entry_cost=5,
        participation_reward=2,
        eligibility_targets=[{"chat_id": -200}],
    )
    app_modules.app.members[(-200, 999)] = _member(ChatMemberStatus.ADMINISTRATOR)
    for user_id in (10, 11):
        status, _ = await service.join_giveaway(
            giveaway["giveaway_id"], user_id=user_id,
            username=None, first_name=str(user_id),
        )
        assert status == "ok"
    app_modules.app.members[(-200, 999)] = _member(ChatMemberStatus.MEMBER)
    with pytest.raises(service.GiveawayServiceError, match="资格核验暂不可用"):
        await service.finish_and_publish_giveaway(giveaway["giveaway_id"], publish=False)
    paused = await database.get_giveaway(giveaway["giveaway_id"])
    assert paused["status"] == database.STATUS_DRAWING
    assert paused["eligibility_error"]
    app_modules.app.members[(-200, 999)] = _member(ChatMemberStatus.ADMINISTRATOR)
    app_modules.app.members[(-200, 11)] = _member(ChatMemberStatus.LEFT)
    completed, _, current = await service.finish_and_publish_giveaway(
        giveaway["giveaway_id"], publish=False
    )
    assert completed is True
    assert [winner["user_id"] for winner in current["winners"]] == [10]
    leaver = await database.get_participant(giveaway["giveaway_id"], 11)
    assert leaver["eligibility_status"] == "ineligible"
    assert (await points.get_point_account(-100, 11))["balance"] == 15


async def test_paused_preselection_draw_can_be_canceled(app_modules):
    database = app_modules.load("wbb.utils.dbgiveaway")
    service = app_modules.load("wbb.services.giveaways")
    giveaway = await database.create_giveaway(
        chat_id=-100, creator_id=1, creator_name="Admin", title="Paused",
        description="", prizes=[{"name": "First", "count": 1}],
        ends_at=datetime.now(UTC) + timedelta(hours=1),
    )
    await database.claim_giveaway_draw(giveaway["giveaway_id"])
    canceled, _, current = await service.cancel_and_refund_giveaway(
        giveaway["giveaway_id"], chat_id=-100, publish=False
    )
    assert canceled is True
    assert current["status"] == database.STATUS_CANCELED


async def test_month_end_schedule_and_materialization_is_idempotent(app_modules, monkeypatch):
    templates = app_modules.load("wbb.services.giveaway_templates")
    database = app_modules.load("wbb.utils.dbgiveaway")

    async def unsupported_history(*_args, **_kwargs):
        raise AssertionError("Bot accounts must not request chat history")
        yield

    monkeypatch.setattr(app_modules.app, "get_chat_history", unsupported_history)
    body = {
        "chat_id": -100, "title": "Monthly", "description": "",
        "prizes": [{"name": "First", "count": 1}],
        "day_of_month": 31, "start_time": "00:00",
        "draw_day_offset": 0, "draw_time": "20:00",
    }
    start, draw = templates.period_schedule(body, 2024, 2)
    assert start.astimezone(templates.TIMEZONE).day == 29
    assert draw - start == timedelta(hours=20)
    start, draw = templates.period_schedule(body, 2025, 2)
    assert start.astimezone(templates.TIMEZONE).day == 28

    now = datetime.now(UTC)
    local = now.astimezone(templates.TIMEZONE)
    body["day_of_month"] = local.day
    body["start_time"] = "00:00"
    body["draw_day_offset"] = 1
    body["draw_time"] = "20:00"
    template = await templates.create_template(body, "Admin")
    await templates.materialize_template(template, now=now)
    await templates.materialize_template(template, now=now)
    items = [item async for item in database.giveawaysdb.find({"template_id": template["template_id"]})]
    assert len(items) == 1
    assert items[0]["period_key"] == local.strftime("%Y-%m")
    assert items[0]["message_id"] == 1
    assert len(app_modules.app.sent_messages) == 1


async def test_announcement_send_failure_retries_same_period(app_modules, monkeypatch):
    templates = app_modules.load("wbb.services.giveaway_templates")
    database = app_modules.load("wbb.utils.dbgiveaway")
    now = datetime.now(UTC)
    local = now.astimezone(templates.TIMEZONE)
    template = await templates.create_template({
        "chat_id": -100, "title": "Retry", "description": "",
        "prizes": [{"name": "First", "count": 1}],
        "day_of_month": local.day, "start_time": "00:00",
        "draw_day_offset": 1, "draw_time": "20:00",
    }, "Admin")
    original_send = app_modules.app.send_message
    failed_once = False

    async def flaky_send(*args, **kwargs):
        nonlocal failed_once
        if not failed_once:
            failed_once = True
            raise RuntimeError("temporary Telegram failure")
        return await original_send(*args, **kwargs)

    monkeypatch.setattr(app_modules.app, "send_message", flaky_send)
    with pytest.raises(RuntimeError, match="temporary Telegram failure"):
        await templates.materialize_template(template, now=now)
    await templates.materialize_template(template, now=now)
    items = [item async for item in database.giveawaysdb.find({
        "template_id": template["template_id"]
    })]
    assert len(items) == 1
    assert items[0]["message_id"] == 1
    assert len(app_modules.app.sent_messages) == 1


async def test_due_unannounced_period_cancels_instead_of_drawing(app_modules):
    database = app_modules.load("wbb.utils.dbgiveaway")
    service = app_modules.load("wbb.services.giveaways")
    giveaway = await database.create_giveaway(
        chat_id=-100, creator_id=1, creator_name="Admin", title="Unannounced",
        description="", prizes=[{"name": "First", "count": 1}],
        ends_at=datetime.now(UTC) + timedelta(hours=1),
        template_id="monthly-template", period_key="2026-09",
    )
    accepted, reason, _ = await service.finish_and_publish_giveaway(
        giveaway["giveaway_id"], publish=False
    )
    assert accepted is False
    assert "报名公告未发布" in reason
    await database.giveawaysdb.update_one(
        {"giveaway_id": giveaway["giveaway_id"]},
        {"$set": {"ends_at": datetime.now(UTC) - timedelta(seconds=1)}},
    )
    due = await database.list_due_giveaways()
    await service.resume_pending_giveaway(due[0])
    canceled = await database.get_giveaway(giveaway["giveaway_id"])
    assert canceled["status"] == database.STATUS_CANCELED
    assert canceled["cancellation_reason"] == "announcement_not_published"
    assert canceled["winners"] == []
    assert not app_modules.app.sent_messages
    assert not app_modules.app.pinned_messages


async def test_due_period_recovers_draw_and_pins_result_once(app_modules, monkeypatch):
    database = app_modules.load("wbb.utils.dbgiveaway")
    service = app_modules.load("wbb.services.giveaways")

    async def unsupported_history(*_args, **_kwargs):
        raise AssertionError("Bot accounts must not request chat history")
        yield

    monkeypatch.setattr(app_modules.app, "get_chat_history", unsupported_history)
    giveaway = await database.create_giveaway(
        chat_id=-100, creator_id=1, creator_name="Admin", title="Catch-up",
        description="", prizes=[{"name": "First", "count": 1}],
        ends_at=datetime.now(UTC) + timedelta(hours=1),
        template_id="monthly-template", period_key="2026-09",
    )
    await database.attach_giveaway_message(giveaway["giveaway_id"], -100, 99)
    await database.giveawaysdb.update_one(
        {"giveaway_id": giveaway["giveaway_id"]},
        {"$set": {"ends_at": datetime.now(UTC) - timedelta(seconds=1)}},
    )
    due = await database.list_due_giveaways()
    assert [item["giveaway_id"] for item in due] == [giveaway["giveaway_id"]]
    await service.resume_pending_giveaway(due[0])
    finished = await database.get_giveaway(giveaway["giveaway_id"])
    assert finished["status"] == database.STATUS_FINISHED
    assert finished["result_pinned"] is True
    assert finished["publication_pending"] is False
    assert app_modules.app.pinned_messages == [(-100, finished["result_message_id"])]
    assert app_modules.app.deleted_messages == [(-100, 99)]
    assert await database.list_due_giveaways() == []


async def test_external_snapshot_parser_and_observation_default(app_modules):
    risk = app_modules.load("wbb.services.external_risk_rules")
    assert risk.parse_entries(
        "yt_spam_domains", b"# comment\nexample.com\nhttps://bad.example/path\nnot a domain\n"
    ) == ["bad.example", "example.com"]
    await risk.ensure_indexes()
    await risk.snapshotsdb.insert_one({
        "snapshot_id": "s1", "bot_id": "primary", "source_id": "yt_spam_domains",
        "kind": "domain", "status": "pending",
    })
    await risk.entriesdb.insert_one({"snapshot_id": "s1", "value": "example.com"})
    await risk.publish_snapshot("s1")
    deleted: list[bool] = []

    async def delete():
        deleted.append(True)

    message = SimpleNamespace(
        id=5, chat=SimpleNamespace(id=-100),
        text="Visit https://sub.example.com/path", caption=None,
        entities=None, caption_entities=None,
        from_user=SimpleNamespace(id=10, is_bot=False), delete=delete,
    )
    await risk.enforce_external(message)
    assert deleted == []
    assert await risk.eventsdb.count_documents({"chat_id": -100}) == 1
    await risk.set_policy(-100, "delete")
    app_modules.app.members[(-100, 999)] = SimpleNamespace(
        status=ChatMemberStatus.ADMINISTRATOR,
        privileges=SimpleNamespace(can_delete_messages=True, can_restrict_members=False),
    )
    await risk.enforce_external(message)
    assert deleted == [True]


async def test_admin_template_api_validates_target_and_exposes_risk_policy(app_modules):
    admin_api = app_modules.load("wbb.admin.api")
    application = admin_api.build_admin_application()
    await application["admin_api"].initialize()
    client = TestClient(TestServer(application), cookie_jar=CookieJar(unsafe=True))
    await client.start_server()
    try:
        login = await client.post(
            "/api/admin/v1/auth/login",
            json={"username": "admin", "password": "qwe0.123456"},
        )
        token = (await login.json())["data"]["csrf_token"]
        changed = await client.put(
            "/api/admin/v1/auth/password",
            headers={"X-CSRF-Token": token},
            json={"current_password": "qwe0.123456", "new_password": "changed-pass-123"},
        )
        token = (await changed.json())["data"]["csrf_token"]
        headers = {"X-CSRF-Token": token}
        app_modules.app.members[(-100, 999)] = _member(ChatMemberStatus.OWNER)
        app_modules.app.members[(-200, 999)] = _member(ChatMemberStatus.MEMBER)
        app_modules.app.chats[-200] = SimpleNamespace(
            id=-200, title="News", username="news", type="channel"
        )
        body = {
            "chat_id": -100, "title": "Monthly", "description": "",
            "prizes": [{"name": "First", "count": 1}],
            "day_of_month": 31, "start_time": "00:00",
            "draw_day_offset": 0, "draw_time": "20:00",
            "eligibility_targets": [{"chat_id": -200}], "confirm": True,
        }
        denied = await client.post(
            "/api/admin/v1/giveaways/templates", headers=headers, json=body
        )
        assert denied.status == 400
        app_modules.app.members[(-200, 999)] = _member(ChatMemberStatus.ADMINISTRATOR)
        created = await client.post(
            "/api/admin/v1/giveaways/templates", headers=headers, json=body
        )
        assert created.status == 201
        template = (await created.json())["data"]
        assert template["eligibility_targets"][0]["title"] == "News"
        listed = await client.get("/api/admin/v1/giveaways/templates")
        assert (await listed.json())["data"]["items"][0]["template_id"] == template["template_id"]
        policy = await client.get("/api/admin/v1/external-risk/chats/-100")
        assert (await policy.json())["data"]["mode"] == "observe"
    finally:
        await client.close()


def test_new_admin_routes_keep_bot_role_permissions(app_modules):
    permissions = app_modules.load("wbb.services.bot_permissions")
    assert permissions.api_permission("POST", "/api/admin/v1/giveaways/templates") == "giveaways.manage"
    assert permissions.api_permission("POST", "/api/admin/v1/external-risk/sources/yt_spam_domains/stage") == "automation.manage"
    assert permissions.api_permission("PATCH", "/api/admin/v1/chats/-100/interaction-settings") == "automation.manage"