from __future__ import annotations from datetime import timedelta import pytest from sqlalchemy import update from app.database.base import utcnow from app.database.models import ( JobStatus, JobType, MembershipMode, PostStatus, ScheduledJob, ) from app.errors import DeliveryError, RateLimitedError, TokenError from app.services.membership_service import effective_channels from tests.helpers import add_channel, create_ready_post, publish_post @pytest.mark.asyncio async def test_token_is_independent_unguessable_and_has_lifecycle(ctx): post = await create_ready_post(ctx) tokens = await ctx.repos.tokens.list_for_post(post.id) assert len(tokens) == 1 raw = tokens[0].token assert len(raw) >= 32 assert raw in (post.video_link or "") assert (post.video_link or "").startswith("https://t.me/smoke_bot?start=") valid = await ctx.tokens.validate(raw) assert valid.valid and valid.reason == "ok" # max uses is consumed atomically and cannot be exceeded await ctx.tokens.set_limits(tokens[0].id, expires_at=None, max_uses=1) first = await ctx.tokens.validate(raw, consume=True) second = await ctx.tokens.validate(raw, consume=True) assert first.valid is True assert second.valid is False and second.reason == "exhausted" # expiry is enforced await ctx.tokens.set_limits( tokens[0].id, expires_at=utcnow() - timedelta(minutes=1), max_uses=None ) expired = await ctx.tokens.validate(raw) assert expired.valid is False and expired.reason == "expired" # regeneration deactivates the old token and returns a fresh link new_link = await ctx.tokens.regenerate(post.id, actor_id=111_000_001) assert new_link != post.video_link old = await ctx.tokens.validate(raw) assert old.valid is False and old.reason == "inactive" latest = await ctx.repos.tokens.latest_for_post(post.id) assert latest is not None and latest.is_active assert (await ctx.tokens.validate(latest.token)).valid # an inactive token can be reactivated from the admin panel path await ctx.tokens.set_active(latest.id, False, 111_000_001) latest = await ctx.repos.tokens.latest_for_post(post.id) assert latest is not None and latest.is_active is False await ctx.tokens.set_active(latest.id, True, 111_000_001) assert (await ctx.tokens.validate(latest.token)).valid @pytest.mark.asyncio async def test_token_rejects_invalid_and_post_archived(ctx): with pytest.raises(TokenError): await ctx.delivery.request( "does-not-exist", telegram_id=1, username=None, display_name=None ) post = await create_ready_post(ctx) await publish_post(ctx, post.id) raw = (await ctx.repos.tokens.active_for_post(post.id)).token # type: ignore[union-attr] await ctx.posts.archive(post.id, 111_000_001) with pytest.raises(TokenError): await ctx.delivery.request(raw, telegram_id=1, username=None, display_name=None) @pytest.mark.asyncio async def test_membership_checks_unconfirmed_and_api_errors_are_never_joined(ctx, gateway): post = await create_ready_post(ctx) channel, _ = await ctx.repos.membership.add_required( chat_id=-100_300_300, title="Required", invite_link="https://t.me/required", chat_type="channel", ) result = await ctx.membership.check_user(post.id, 555) assert result.all_joined is False assert len(result.unconfirmed) == 1 assert result.unconfirmed[0].joined is False assert result.has_errors is False gateway.member_errors.add(channel.chat_id) result = await ctx.membership.check_user(post.id, 555) assert result.all_joined is False assert result.has_errors is True assert result.unconfirmed[0].joined is None assert result.unconfirmed[0].error gateway.member_errors.clear() gateway.memberships[(channel.chat_id, 555)] = "member" result = await ctx.membership.check_user(post.id, 555) assert result.all_joined is True assert result.unconfirmed == [] def test_effective_channels_priority_and_overrides(): from app.database.models import Post, RequiredChannel global_channel = RequiredChannel( id=1, chat_id=-1, title="Global", invite_link="https://t.me/a", is_active=True, is_required_globally=True, sort_order=1, ) optional_channel = RequiredChannel( id=2, chat_id=-2, title="Optional", invite_link="https://t.me/b", is_active=True, is_required_globally=False, sort_order=2, ) post = Post(id=1, media_asset_id=1, membership_mode=MembershipMode.INHERIT) assert [c.id for c in effective_channels(post, [global_channel, optional_channel], {})] == [1] post.membership_mode = MembershipMode.ENFORCE assert [c.id for c in effective_channels(post, [global_channel, optional_channel], {})] == [1, 2] post.membership_mode = MembershipMode.DISABLE assert effective_channels(post, [global_channel, optional_channel], {}) == [] post.membership_mode = MembershipMode.INHERIT assert [c.id for c in effective_channels(post, [global_channel, optional_channel], {1: False})] == [] assert [c.id for c in effective_channels(post, [global_channel, optional_channel], {1: False, 2: True})] == [2] @pytest.mark.asyncio async def test_membership_gate_consumes_token_only_after_success_and_delivers(ctx, gateway): await ctx.app_settings.set("rate_limit_min_interval_seconds", 0) post = await create_ready_post(ctx) channel, _ = await ctx.repos.membership.add_required( chat_id=-100_400_400, title="Required", invite_link="https://t.me/required", chat_type="channel", ) await publish_post(ctx, post.id) token = await ctx.repos.tokens.active_for_post(post.id) assert token is not None raw = token.token outcome = await ctx.delivery.request( raw, telegram_id=777, username="u", display_name="User" ) assert outcome.kind == "membership" assert outcome.membership is not None and not outcome.membership.all_joined assert await ctx.repos.tokens.get_by_token(raw) is not None assert (await ctx.repos.tokens.get_by_token(raw)).use_count == 0 # type: ignore[union-attr] assert await ctx.repos.deliveries.count() == 0 gateway.memberships[(channel.chat_id, 777)] = "member" outcome = await ctx.delivery.request( raw, telegram_id=777, username="u", display_name="User" ) assert outcome.kind == "delivered" assert outcome.delivery is not None assert await ctx.repos.deliveries.count() == 1 token = await ctx.repos.tokens.get_by_token(raw) assert token is not None and token.use_count == 1 assert gateway.sent_videos[-1]["chat_id"] == 777 @pytest.mark.asyncio async def test_auto_delete_creates_separate_durable_jobs_and_recovers(ctx, gateway): await ctx.app_settings.set("auto_delete_enabled", True) await ctx.app_settings.set("user_delete_seconds", 30) await ctx.app_settings.set("channel_delete_seconds", 60) post = await create_ready_post(ctx) await publish_post(ctx, post.id) token = await ctx.repos.tokens.active_for_post(post.id) assert token is not None outcome = await ctx.delivery.request( token.token, telegram_id=888, username=None, display_name="User" ) assert outcome.kind == "delivered" jobs = await ctx.repos.jobs.list_recent(20) types = {job.job_type for job in jobs} assert JobType.DELETE_USER_MESSAGE in types assert JobType.DELETE_CHANNEL_MESSAGE in types assert all(job.status == JobStatus.PENDING for job in jobs) # Make both jobs due without waiting in a handler or test sleep. async with ctx.db.session() as session: await session.execute( update(ScheduledJob).values(run_at=utcnow() - timedelta(seconds=1)) ) processed = await ctx.scheduler.run_once(10) assert processed >= 2 assert len(gateway.deleted) >= 2 assert any(chat_id == 888 for chat_id, _ in gateway.deleted) assert any(chat_id != 888 for chat_id, _ in gateway.deleted) delivery = await ctx.repos.deliveries.get(outcome.delivery.id) # type: ignore[union-attr] assert delivery is not None and delivery.deleted_at is not None @pytest.mark.asyncio async def test_rate_limit_and_blocked_user(ctx): post = await create_ready_post(ctx) await publish_post(ctx, post.id) token = await ctx.repos.tokens.active_for_post(post.id) assert token is not None await ctx.app_settings.set("rate_limit_requests", 1) await ctx.app_settings.set("rate_limit_window_seconds", 60) first = await ctx.delivery.request( token.token, telegram_id=999, username=None, display_name="U" ) assert first.kind == "delivered" with pytest.raises(RateLimitedError): await ctx.delivery.request( token.token, telegram_id=999, username=None, display_name="U" ) await ctx.repos.users.set_blocked(999, True, "test") with pytest.raises(DeliveryError): await ctx.delivery.request( token.token, telegram_id=999, username=None, display_name="U" ) @pytest.mark.asyncio async def test_publishing_sets_published_status_and_link(ctx): channel = await add_channel(ctx) post = await create_ready_post(ctx, channel_ids=[channel.id]) assert post.status == PostStatus.READY result = await ctx.publishing.publish(post.id, actor_id=111_000_001) assert result.published_count == 1 refreshed = await ctx.repos.posts.get(post.id) assert refreshed is not None assert refreshed.status == PostStatus.PUBLISHED assert refreshed.published_at is not None assert refreshed.video_link destination = (await ctx.repos.posts.destinations(post.id))[0] assert destination.telegram_message_id is not None