from __future__ import annotations import logging from datetime import UTC, datetime import pytest from app.database.models import ( DestinationStatus, JobStatus, JobType, PostStatus, ) from app.errors import PublishError from app.gateways.base import ChatInfo from app.logging_config import SecretRedactionFilter from tests.helpers import add_channel, create_ready_post, publish_post @pytest.mark.asyncio async def test_multiple_posts_get_independent_media_rows(ctx): first = await create_ready_post(ctx, title="First") second = await create_ready_post(ctx, title="Second") assert first.id != second.id assert first.media_asset_id != second.media_asset_id assert first.media_asset.file_unique_id != second.media_asset.file_unique_id @pytest.mark.asyncio async def test_post_lifecycle_pause_resume_archive_republish_clone_delete(ctx, gateway): channel = await add_channel(ctx, -100_601_601) post = await create_ready_post(ctx, channel_ids=[channel.id]) await ctx.posts.pause(post.id, 111_000_001) paused = await ctx.repos.posts.get(post.id) assert paused is not None and paused.status == PostStatus.PAUSED with pytest.raises(PublishError): await ctx.posts.publish_now(post.id, 111_000_001) await ctx.posts.resume(post.id, 111_000_001) first_result = await ctx.posts.publish_now(post.id, 111_000_001) assert first_result.status == PostStatus.PUBLISHED first_destination = (await ctx.repos.posts.destinations(post.id))[0] first_message_id = first_destination.telegram_message_id assert first_message_id is not None await ctx.posts.archive(post.id, 111_000_001) archived = await ctx.repos.posts.get(post.id) assert archived is not None and archived.status == PostStatus.ARCHIVED second_result = await ctx.posts.republish(post.id, 111_000_001) assert second_result.status == PostStatus.PUBLISHED second_destination = (await ctx.repos.posts.destinations(post.id))[0] assert second_destination.status == DestinationStatus.PUBLISHED assert second_destination.telegram_message_id != first_message_id assert gateway.send_attempts == 2 clone = await ctx.posts.clone(post.id, 111_000_001) assert clone.status == PostStatus.DRAFT clone_destinations = await ctx.repos.posts.destinations(clone.id) assert len(clone_destinations) == 1 assert clone_destinations[0].channel_id == channel.id await ctx.posts.delete(post.id, 111_000_001) assert await ctx.repos.posts.get(post.id) is None assert await ctx.repos.posts.get(clone.id) is not None @pytest.mark.asyncio async def test_republish_schedules_new_channel_delete_job_and_delete_cancels_jobs(ctx): await ctx.app_settings.set("auto_delete_enabled", True) await ctx.app_settings.set("channel_delete_seconds", 60) channel = await add_channel(ctx, -100_602_602) post = await create_ready_post(ctx, channel_ids=[channel.id]) await publish_post(ctx, post.id) await ctx.posts.republish(post.id, 111_000_001) jobs = await ctx.repos.jobs.list_recent(20) channel_jobs = [job for job in jobs if job.job_type == JobType.DELETE_CHANNEL_MESSAGE] assert len(channel_jobs) == 2 assert all(job.status == JobStatus.PENDING for job in channel_jobs) await ctx.posts.delete(post.id, 111_000_001) jobs = await ctx.repos.jobs.list_recent(20) assert jobs assert all(job.status == JobStatus.CANCELLED for job in jobs) @pytest.mark.asyncio async def test_global_and_post_caption_priority_and_protect_content(ctx, gateway): await ctx.app_settings.set("caption_template", "GLOBAL {title} {video_link}") await ctx.app_settings.set("brand_signature", "BRAND") await ctx.app_settings.set("protect_content", True) post = await create_ready_post(ctx, title="Title", description="Description") await ctx.posts.set_caption_override(post.id, "POST {title} {description}") await ctx.posts.set_protect_content(post.id, False) await publish_post(ctx, post.id) sent = gateway.sent_videos[-1] assert sent["protect_content"] is False assert "POST Title Description" in sent["caption"] assert "BRAND" in sent["caption"] @pytest.mark.asyncio async def test_caption_publish_date_uses_display_timezone(ctx): await ctx.app_settings.set("timezone", "Asia/Tehran") await ctx.app_settings.set("caption_template", "{publish_date}") post = await create_ready_post(ctx) caption = await ctx.app_settings.build_caption( post, publish_date=datetime(2026, 1, 1, 12, 0, tzinfo=UTC), ) assert caption == "2026-01-01 15:30" @pytest.mark.asyncio async def test_post_specific_caption_extras_and_link_template(ctx): await ctx.app_settings.set("caption_template", "{title}|{video_link}") await ctx.app_settings.set("caption_prefix", "GLOBAL-PREFIX") await ctx.app_settings.set("caption_suffix", "GLOBAL-SUFFIX") await ctx.app_settings.set("brand_signature", "GLOBAL-BRAND") post = await create_ready_post(ctx, title="Title") await ctx.posts.set_caption_prefix_override(post.id, "POST-PREFIX") await ctx.posts.set_caption_suffix_override(post.id, "POST-SUFFIX") await ctx.posts.set_signature_override(post.id, "POST-BRAND") await ctx.posts.set_video_link_template_override( post.id, "https://t.me/{bot_username}?start={token}&source=post" ) link = await ctx.tokens.regenerate(post.id, actor_id=111_000_001) refreshed = await ctx.repos.posts.get(post.id) assert refreshed is not None assert "source=post" in link assert refreshed.video_link == link caption = await ctx.app_settings.build_caption(refreshed) assert caption.startswith("POST-PREFIX") assert "POST-SUFFIX" in caption assert "POST-BRAND" in caption assert "GLOBAL-" not in caption @pytest.mark.asyncio async def test_channel_add_and_permission_check(ctx, gateway): chat_id = -100_603_603 gateway.chats[chat_id] = ChatInfo( chat_id=chat_id, title="Destination", username="destination", chat_type="channel", ) gateway.bot_admin.add(chat_id) channel, created = await ctx.channels.add(chat_id, 111_000_001) assert created is True assert channel.can_post is True assert await ctx.channels.check(channel.id) is True gateway.bot_admin.clear() assert await ctx.channels.check(channel.id) is False @pytest.mark.asyncio async def test_membership_channel_add_and_bot_access_check(ctx, gateway): chat_id = -100_604_604 gateway.chats[chat_id] = ChatInfo( chat_id=chat_id, title="Required", username="required_channel", chat_type="channel", ) gateway.bot_admin.add(chat_id) channel = await ctx.membership.add_channel( chat_id, title=None, invite_link=None, actor_id=111_000_001, ) assert channel.invite_link == "https://t.me/required_channel" assert await ctx.membership.check_bot_access(channel) is True gateway.bot_admin.clear() assert await ctx.membership.check_bot_access(channel) is False def test_log_redaction_hides_bot_token_and_deep_link(): token = "123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghi" link = "https://t.me/smoke_bot?start=very-secret-token" record = logging.LogRecord( "test", logging.INFO, __file__, 1, "token=%s link=%s", (token, link), None, ) SecretRedactionFilter().filter(record) message = record.getMessage() assert token not in message assert "very-secret-token" not in message assert "[REDACTED_BOT_TOKEN]" in message assert "start=[REDACTED]" in message