import os import tempfile import unittest from pathlib import Path from bot import App, Config, Store, APIError, caption, duration_display, parse_duration, _visible as _visible_for_test class FakeAPI: def __init__(self): self.calls = [] self.fail = None self.fail_error = None self.member_status = 'member' def call(self, method, **data): self.calls.append((method, data)) if self.fail_error is not None and (self.fail is None or self.fail == method): raise self.fail_error if self.fail == method: raise APIError('transport', uncertain=True) if method == 'getChatMember': return {'status': self.member_status} if method == 'sendMediaGroup': return [{'message_id': len(self.calls) + index, 'chat': {'id': -100123}} for index in range(len(data.get('media', [])))] return {'message_id': len(self.calls), 'chat': {'id': -100123}} class FlowTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db = Store(Path(self.tmp.name) / 'test.db') self.api = FakeAPI() self.app = App(Config('123:token', {1}, '@femonascene', '-100999', str(Path(self.tmp.name) / 'test.db')), self.db, self.api, 'FemonaTestBot') self.uid = 0 def tearDown(self): self.app.close() self.db.conn.close() self.tmp.cleanup() def drain(self): self.app.drain() def message(self, text='', user=1, **fields): self.uid += 1 self.app.process({'update_id': self.uid, 'message': {'message_id': self.uid, 'from': {'id': user}, 'chat': {'id': user, 'type': 'private'}, 'text': text, **fields}}) def click(self, data, user=1): self.uid += 1 self.app.process({'update_id': self.uid, 'callback_query': {'id': str(self.uid), 'from': {'id': user}, 'data': data, 'message': {'message_id': 99, 'chat': {'id': user, 'type': 'private'}}}}) def new_id(self): return self.db.one('SELECT * FROM movies ORDER BY id DESC LIMIT 1')['id'] def draft(self): """Admin types only title/original/imdb/summary/duration; taxonomy picked with buttons.""" self.message('/new') for value in ['فیلم آزمایشی', 'Test Film', '7.5', 'داستان آزمایشی', '02:28:03']: self.message(value) self.message(photo=[{'file_id': 'poster', 'file_size': 100}]) self.message('/skip') mid = self.new_id() self.click(f'catsel:{mid}:1') self.click(f'gentgl:{mid}:1') self.click(f'gendone:{mid}') self.click(f'yrsel:{mid}:2025') self.click(f'actskip:{mid}') self.message('کارگردان آزمایشی') self.message('1080p | زیرنویس فارسی چسبیده') self.message(document={'file_id': 'movie-file', 'file_unique_id': 'unique', 'file_size': 1700000000, 'file_name': 'film.mkv'}) return self.db.one('SELECT * FROM movies WHERE id=?', mid) def test_wizard_stages_files_without_server_download(self): movie = self.draft() self.assertEqual(movie['title'], 'فیلم آزمایشی') self.assertEqual(movie['category'], 'فیلم سینمایی') self.assertEqual(movie['imdb'], '7.5') self.assertEqual(movie['year'], '2025') self.assertEqual(movie['duration'], '2:28:03') self.assertEqual(movie['director'], 'کارگردان آزمایشی') self.assertEqual(self.db.one('SELECT genre_id FROM movie_genres WHERE movie_id=?', movie['id'])['genre_id'], 1) variant = self.db.one('SELECT * FROM variants') self.assertEqual(variant['size'], 1700000000) self.assertEqual(variant['file_id'], 'movie-file') # v3: files are staged by file_id; the archive copy happens later, in one album. self.assertEqual(variant['archive_message'], 0) self.assertEqual(movie['poster_archive'], 0) self.assertFalse(any(m == 'copyMessage' for m, _ in self.api.calls)) self.assertFalse(any(m == 'getFile' for m, _ in self.api.calls)) def test_batch_send_to_storage_as_one_album(self): movie = self.draft() mid = movie['id'] self.click(f"traileredit:{mid}") self.message(video={'file_id': 'trailer-file', 'file_size': 5000}) self.click(f'archnow:{mid}') groups = [d for m, d in self.api.calls if m == 'sendMediaGroup' and d['chat_id'] == '-100999'] self.assertEqual(len(groups), 1) media_types = [item['type'] for item in groups[0]['media']] self.assertEqual(media_types, ['photo', 'video']) self.assertEqual(groups[0]['media'][1]['media'], 'trailer-file') updated = self.db.movie(mid) self.assertNotEqual(updated['poster_archive'], 0) self.assertNotEqual(updated['trailer_archive'], 0) self.assertNotEqual(self.db.one('SELECT * FROM variants')['archive_message'], 0) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM variants WHERE archive_message=0')['n'], 0) def test_batch_send_failure_keeps_retry_pending(self): movie = self.draft() mid = movie['id'] self.click(f"traileredit:{mid}") self.message(video={'file_id': 'trailer-file', 'file_size': 5000}) self.api.fail = 'sendMediaGroup' self.click(f'archnow:{mid}') # The album (cover+trailer) failed; the lone document still went through individually. self.assertEqual(self.db.movie(mid)['poster_archive'], 0) self.assertEqual(self.db.movie(mid)['trailer_archive'], 0) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM variants WHERE archive_message=0')['n'], 0) self.api.fail = None self.click(f'archnow:{mid}') self.assertGreater(self.db.movie(mid)['poster_archive'], 0) self.assertGreater(self.db.movie(mid)['trailer_archive'], 0) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM variants WHERE archive_message=0')['n'], 0) def test_unauthorized_admin_callback_does_not_publish(self): movie = self.draft() self.click(f"publish:{movie['id']}", user=2) self.assertEqual(self.db.movie(movie['id'])['status'], 'draft') self.assertFalse(any(d.get('chat_id') == '@femonascene' for _, d in self.api.calls)) def test_publish_once_and_deep_link(self): movie = self.draft() for _ in range(2): self.click(f"publish:{movie['id']}") posts = [d for m, d in self.api.calls if m == 'sendPhoto' and d['chat_id'] == '@femonascene'] self.assertEqual(len(posts), 1) markup = posts[0]['reply_markup']['inline_keyboard'][0] self.assertIn('start=m_', markup[0]['url']) self.assertIn('start=panel', markup[1]['url']) self.assertEqual(self.db.movie(movie['id'])['status'], 'published') def test_post_caption_includes_taxonomy(self): movie = self.draft() self.click(f"publish:{movie['id']}") post = next(d for m, d in self.api.calls if m == 'sendPhoto' and d['chat_id'] == '@femonascene') self.assertIn('امتیاز IMDb', post['caption']) self.assertIn('دسته‌بندی', post['caption']) self.assertIn('ژانر', post['caption']) self.assertIn('اکشن', post['caption']) self.assertIn('مدت‌زمان: ۲:۲۸:۰۳ (۱۴۸ دقیقه)', post['caption']) self.assertIn('کارگردان', post['caption']) def test_channel_post_carries_bot_link_in_caption_and_buttons(self): movie = self.draft() self.click(f"publish:{movie['id']}") post = next(d for m, d in self.api.calls if m == 'sendPhoto' and d['chat_id'] == '@femonascene') # Visible text link inside the caption (works even if a client hides buttons) self.assertIn('انتخاب کیفیت و دانلود از ربات', post['caption']) self.assertIn(f"start=m_{movie['token']}", post['caption']) self.assertIn('#فیلم_سینمایی', post['caption']) self.assertIn('#اکشن', post['caption']) self.assertIn('🆔 @femonascene', post['caption']) def test_latin_title_is_rtl_right_aligned(self): movie = self.draft() text = self.app.movie_card_text(movie) lines = text.split('\n') label_index = lines.index('🌐 نام لاتین:') value_line = lines[label_index + 1] self.assertTrue(value_line.startswith('\u200f')) self.assertEqual(value_line[1:], 'Test Film') def test_draft_and_hidden_files_are_not_public(self): movie = self.draft() variant = self.db.one('SELECT * FROM variants') for status in ['draft', 'hidden']: self.db.run('UPDATE movies SET status=? WHERE id=?', status, movie['id']) self.click(f"file:{variant['id']}", user=2) self.drain() self.assertFalse(any(m == 'sendDocument' for m, _ in self.api.calls)) def test_published_file_and_metrics(self): movie = self.draft() self.click(f"publish:{movie['id']}") variant = self.db.one('SELECT * FROM variants') self.click(f"file:{variant['id']}", user=2) self.drain() sends = [d for m, d in self.api.calls if m == 'sendDocument'] self.assertEqual(sends[-1]['document'], 'movie-file') self.assertEqual(self.db.one("SELECT COUNT(*) n FROM events WHERE kind='sent'")['n'], 1) def test_ambiguous_publish_requires_recovery_not_auto_retry(self): movie = self.draft() self.api.fail = 'sendPhoto' self.click(f"publish:{movie['id']}") self.assertEqual(self.db.movie(movie['id'])['status'], 'uncertain') self.api.fail = None self.click(f"publish:{movie['id']}") posts = [d for m, d in self.api.calls if m == 'sendPhoto' and d['chat_id'] == '@femonascene'] self.assertEqual(len(posts), 1) def test_update_redelivery_does_not_duplicate_creation(self): update = {'update_id': 800, 'message': {'message_id': 1, 'from': {'id': 1}, 'chat': {'id': 1, 'type': 'private'}, 'text': '/new'}} self.app.process(update) self.app.process(update) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM movies')['n'], 1) def test_missing_file_prevents_publication(self): self.message('/new') movie = self.db.one('SELECT * FROM movies') self.click(f"publish:{movie['id']}") self.assertEqual(self.db.movie(movie['id'])['status'], 'draft') def test_group_admin_commands_ignored(self): self.app.process({'update_id': 70, 'message': {'from': {'id': 1}, 'chat': {'id': -100, 'type': 'supergroup'}, 'text': '/new'}}) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM movies')['n'], 0) def test_search_normalizes_persian_and_hides_drafts(self): movie = self.draft() self.db.run("UPDATE movies SET title='کتاب یک', search='کتاب یک test', status='published' WHERE id=?", movie['id']) self.message('كتاب يك', user=2) self.assertIn('کتاب یک', str(self.api.calls[-1])) self.db.run("UPDATE movies SET status='hidden' WHERE id=?", movie['id']) self.message('كتاب يك', user=2) self.assertIn('پیدا نشد', str(self.api.calls[-1])) def test_search_finds_genre_and_actor_names(self): movie = self.draft() self.click(f"publish:{movie['id']}") self.db.run("INSERT OR IGNORE INTO actors(name,search) VALUES ('Leonardo DiCaprio','leonardo dicaprio')") aid = self.db.one("SELECT id FROM actors WHERE search='leonardo dicaprio'")['id'] self.db.run('INSERT OR IGNORE INTO movie_actors(movie_id,actor_id) VALUES (?,?)', movie['id'], aid) self.app.refresh_search(movie['id']) self.message('dicaprio', user=2) self.assertIn('فیلم آزمایشی', str(self.api.calls[-1])) def test_search_finds_director(self): movie = self.draft() self.click(f"publish:{movie['id']}") self.message('کارگردان آزمایشی', user=2) self.assertIn('فیلم آزمایشی', str(self.api.calls[-1])) def test_session_survives_restart(self): self.message('/new') self.message('نام') self.app = App(self.app.cfg, self.db, self.api, 'FemonaTestBot') self.message('Original') self.assertEqual(self.db.one('SELECT * FROM movies')['original'], 'Original') def test_imdb_validation(self): self.message('/new') self.message('نام') self.message('Original') self.message('15') self.assertEqual(self.db.session(1)['step'], 'imdb') self.message('8.5') self.assertEqual(self.db.session(1)['step'], 'summary') def test_caption_escapes_markup(self): movie = self.draft() movie['title'] = 'Fake' self.assertIn('<b>', caption(movie)) def test_report_is_saved_without_sending_to_others(self): movie = self.draft() self.click(f"publish:{movie['id']}") before = len(self.api.calls) self.click(f"report:{movie['id']}", user=2) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM reports')['n'], 1) self.assertFalse(any(d.get('chat_id') == 1 for _, d in self.api.calls[before:])) def test_duplicate_variant_rejected(self): movie = self.draft() self.message('1080p | زیرنویس فارسی چسبیده') self.message(document={'file_id': 'other-file', 'file_size': 100}) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM variants')['n'], 1) def test_replace_file_preserves_download_button(self): movie = self.draft() self.click(f"publish:{movie['id']}") variant = self.db.one('SELECT * FROM variants') self.click(f"replace:{variant['id']}") self.message(document={'file_id': 'corrected', 'file_size': 400}) self.click(f"file:{variant['id']}", user=2) self.drain() sends = [d for m, d in self.api.calls if m == 'sendDocument'] self.assertEqual(sends[-1]['document'], 'corrected') def test_untrusted_replacement_callback_denied(self): self.draft() variant = self.db.one('SELECT * FROM variants') self.click(f"replace:{variant['id']}", user=2) self.message(user=2, document={'file_id': 'malicious', 'file_size': 400}) self.assertEqual(self.db.one('SELECT * FROM variants')['file_id'], 'movie-file') def test_last_published_variant_cannot_be_removed(self): movie = self.draft() self.click(f"publish:{movie['id']}") variant = self.db.one('SELECT * FROM variants') self.click(f"remove:{variant['id']}") self.assertIsNotNone(self.db.one('SELECT * FROM variants')) def test_batch_failure_is_not_counted_as_storage_complete(self): self.message('/new') for value in ['نام', 'Name', '/skip', 'خلاصه', '2:28']: self.message(value) self.message(photo=[{'file_id': 'poster'}]) self.message('/skip') mid = self.new_id() self.click(f'catsel:{mid}:1') self.click(f'gendone:{mid}') self.click(f'yrsel:{mid}:2025') self.click(f'actskip:{mid}') self.message('/skip') self.message('720p') self.message(document={'file_id': 'file', 'file_size': 100}) self.assertEqual(self.db.session(1).get('step'), 'quality') self.assertEqual(self.db.one('SELECT archive_message FROM variants WHERE movie_id=?', mid)['archive_message'], 0) self.assertEqual(self.db.movie(mid)['duration'], '2:28:00') def test_manage_panel_offers_one_click_storage_send(self): movie = self.draft() self.click(f"manage:{movie['id']}") last = self.api.calls[-1][1] self.assertIn('ارسال یکجا به استوریج', str(last.get('reply_markup'))) self.assertIn('استوریج: ارسال نشده', last['text']) self.click(f'archnow:{movie["id"]}') self.assertEqual(self.db.one('SELECT COUNT(*) n FROM variants WHERE archive_message=0')['n'], 0) def test_sql_search_is_literal(self): self.draft() self.message("%' OR 1=1 --", user=2) self.assertIn('پیدا نشد', str(self.api.calls[-1])) def test_hide_blocks_old_download_button_and_restore_reenables(self): movie = self.draft() self.click(f"publish:{movie['id']}") variant = self.db.one('SELECT * FROM variants') self.click(f"hide:{movie['id']}") self.click(f"file:{variant['id']}", user=2) self.assertFalse(any(m == 'sendDocument' for m, _ in self.api.calls)) self.click(f"restore:{movie['id']}") self.click(f"file:{variant['id']}", user=2) self.drain() self.assertTrue(any(m == 'sendDocument' for m, _ in self.api.calls)) def test_backup_is_consistent(self): movie = self.draft() target = Path(self.tmp.name) / 'backup.db' self.db.backup(target) restored = Store(target) self.assertEqual(restored.movie(movie['id'])['title'], movie['title']) self.assertEqual(restored.one('SELECT * FROM variants')['file_id'], 'movie-file') restored.conn.close() def test_removed_variant_id_is_never_reassigned_to_other_file(self): movie = self.draft() old_id = self.db.one('SELECT * FROM variants')['id'] self.click(f'remove:{old_id}') self.click(f"add:{movie['id']}") self.message('720p') self.message(document={'file_id': 'new-version', 'file_size': 123}) self.assertNotEqual(self.db.one('SELECT * FROM variants')['id'], old_id) def test_unicode_caption_stays_in_telegram_limit(self): self.message('/new') self.message('🎬' * 100) self.assertEqual(self.db.session(1)['step'], 'title') def test_preview_does_not_publish(self): movie = self.draft() self.click(f"preview:{movie['id']}") self.assertEqual(self.db.movie(movie['id'])['status'], 'draft') self.assertFalse(any(d.get('chat_id') == '@femonascene' for _, d in self.api.calls)) # ---------- v3.0.0 additions ---------- def test_multi_channel_gate_add_toggle_remove(self): movie = self.draft() self.click(f"publish:{movie['id']}") self.click('gates') self.assertIn('عضویت اجباری', self.api.calls[-1][1]['text']) self.click('gateadd') self.api.member_status = 'administrator' self.message(forward_from_chat={'id': -100555, 'type': 'channel', 'title': 'کانال دوم', 'username': 'secondch'}, text='') self.assertEqual(self.db.one("SELECT COUNT(*) n FROM gate_channels WHERE chat_id='-100555'")['n'], 1) self.api.member_status = 'left' self.click(f"file:{self.db.one('SELECT * FROM variants')['id']}", user=2) rows = self.api.calls[-1][1]['reply_markup']['inline_keyboard'] joined = [b['text'] for row in rows for b in row] self.assertTrue(any('کانال دوم' in text for text in joined)) self.assertTrue(any('بررسی مجدد عضویت' in text for text in joined)) self.api.member_status = 'member' self.click('gatecheck', user=2) self.assertIn('تأیید شد', self.api.calls[-1][1]['text']) gid = self.db.one("SELECT id FROM gate_channels WHERE chat_id='-100555'")['id'] self.click(f'gaterm:{gid}') self.assertEqual(self.db.one("SELECT COUNT(*) n FROM gate_channels WHERE chat_id='-100555'")['n'], 0) def test_gate_toggle_off_lets_everyone_download(self): movie = self.draft() self.click(f"publish:{movie['id']}") variant = self.db.one('SELECT * FROM variants') self.api.member_status = 'left' self.click(f"file:{variant['id']}", user=2) self.assertFalse(any(m == 'sendDocument' for m, _ in self.api.calls)) self.click('gatetoggle') self.assertFalse(self.app.gate_enabled()) self.click(f"file:{variant['id']}", user=2) self.drain() self.assertTrue(any(m == 'sendDocument' for m, _ in self.api.calls)) def test_ads_flow_and_rotation_in_delivery_and_channel(self): movie = self.draft() self.click(f"publish:{movie['id']}") self.click('adadd') self.message('کانال حامی ما را دنبال کنید') self.message('حامی ما | https://t.me/sponsor') self.assertEqual(self.db.one('SELECT COUNT(*) n FROM ads')['n'], 1) variant = self.db.one('SELECT * FROM variants') self.click(f"file:{variant['id']}", user=2) self.drain() send = [d for m, d in self.api.calls if m == 'sendDocument'][-1] self.assertIn('کانال حامی ما', send['caption']) self.assertIn('https://t.me/sponsor', send['caption']) self.click(f"syncpost:{movie['id']}") edit = [d for m, d in self.api.calls if m == 'editMessageCaption' and d.get('chat_id') == '@femonascene'][-1] self.assertIn('کانال حامی ما', edit['caption']) self.api.calls.clear() self.click(f"syncpost:{movie['id']}") edit = next(d for m, d in self.api.calls if m == 'editMessageCaption') markup = edit.get('reply_markup', {}).get('inline_keyboard', []) self.assertTrue(any(row and row[0].get('url') == 'https://t.me/sponsor' for row in markup)) def test_ad_toggle_and_bot_switch_off(self): movie = self.draft() self.click('adadd') self.message('تبلیغ آزمایشی') self.message('/skip') self.click('adbot') self.assertFalse(self.app.ads_flag('bot')) self.click(f"publish:{movie['id']}") variant = self.db.one('SELECT * FROM variants') self.click(f"file:{variant['id']}", user=2) self.drain() send = [d for m, d in self.api.calls if m == 'sendDocument'][-1] self.assertNotIn('تبلیغ آزمایشی', send['caption']) self.assertTrue(self.app.ads_flag('channel')) def test_broadcast_flow_confirms_before_sending(self): self.message('/start', user=2) self.message('/start', user=3) self.click('bc') self.message('اطلاع‌رسانی ویژه') self.assertEqual(self.db.session(1)['step'], 'bcconfirm') self.click('bcyes') self.drain() copies = [d for m, d in self.api.calls if m == 'copyMessage'] targets = {d['chat_id'] for d in copies} self.assertEqual(targets, {1, 2, 3}) def test_download_history_in_profile(self): movie = self.draft() self.click(f"publish:{movie['id']}") variant = self.db.one('SELECT * FROM variants') self.click(f"file:{variant['id']}", user=2) self.drain() self.assertEqual(self.db.one('SELECT COUNT(*) n FROM downloads WHERE user_id=2')['n'], 1) self.click('profile', user=2) profile_call = self.api.calls[-1][1] self.assertIn('دانلودهای من', profile_call['text']) self.assertIn('دانلودهای من (۱)', str(profile_call.get('reply_markup'))) self.click('hist:0', user=2) self.assertIn('فیلم آزمایشی', str(self.api.calls[-1])) def test_users_registry_records_private_chatters(self): self.message('/start', user=2) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM users')['n'], 1) self.message('/help', user=3) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM users')['n'], 2) def test_caption_long_summary_is_truncated_to_telegram_limit(self): movie = {'title': 'فیلم', 'original': 'Movie', 'year': '2025', 'duration': '2:00:00', 'summary': 'متن آزمایشی ' * 400} text = caption(movie, '@testch', {'deep_url': 'https://t.me/b?start=m_x', 'tags': '#فیلم #اکشن'}) self.assertLessEqual(_visible_for_test(text), 1000) self.assertIn('…', text) self.assertIn('#فیلم', text) def test_random_movie_and_sync_post_button(self): movie = self.draft() self.click(f"publish:{movie['id']}") self.click('random', user=2) self.assertIn('فیلم آزمایشی', str(self.api.calls[-1])) self.click(f"syncpost:{movie['id']}") self.assertTrue(any(m == 'editMessageCaption' and d.get('chat_id') == '@femonascene' for m, d in self.api.calls)) def test_admin_menu_buttons_are_admin_only(self): self.click('gates', user=2) self.assertIn('مدیر', self.api.calls[-1][1]['text']) self.click('gates', user=1) self.assertIn('عضویت اجباری', self.api.calls[-1][1]['text']) self.click('admenu', user=1) self.assertIn('تبلیغات', self.api.calls[-1][1]['text']) def test_failed_delivery_is_not_counted_as_success(self): movie = self.draft() self.click(f"publish:{movie['id']}") variant = self.db.one('SELECT * FROM variants') self.api.fail = 'sendDocument' self.click(f"file:{variant['id']}", user=2) self.drain() self.assertEqual(self.db.one("SELECT COUNT(*) n FROM events WHERE kind='sent'")['n'], 0) self.assertEqual(self.db.one("SELECT COUNT(*) n FROM events WHERE kind='failed'")['n'], 1) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM downloads')['n'], 0) def test_join_gate_blocks_nonmember_and_allows_member(self): movie = self.draft() self.click(f"publish:{movie['id']}") variant = self.db.one('SELECT * FROM variants') self.app.cfg.force_sub = True self.api.member_status = 'left' self.click(f"file:{variant['id']}", user=2) self.drain() self.assertFalse(any(m == 'sendDocument' for m, _ in self.api.calls)) self.assertTrue(any(d.get('reply_markup') and 'عضویت در کانال' in str(d.get('reply_markup')) for m, d in self.api.calls if m == 'sendMessage')) self.api.member_status = 'member' self.click(f"file:{variant['id']}", user=3) self.drain() self.assertTrue(any(m == 'sendDocument' for m, _ in self.api.calls)) def test_caption_uses_configured_channel(self): movie = {'title': 'T', 'original': 'O', 'year': '2025', 'genre': 'G', 'duration': '90', 'summary': 'S'} self.assertIn('@testch', caption(movie, '@testch')) self.assertNotIn('@', caption(movie, '')) def test_sync_post_ignores_not_modified(self): movie = self.draft() self.click(f"publish:{movie['id']}") self.api.fail_error = APIError('Bad Request: message is not modified', code=400) before = len(self.api.calls) self.app.sync_post(1, movie['id']) after = self.api.calls[before:] self.assertTrue(any(m == 'editMessageCaption' for m, _ in after)) self.assertFalse(any(m == 'sendMessage' and 'تأیید نشد' in d.get('text', '') for m, d in after)) def test_poster_is_staged_without_immediate_archive(self): self.message('/new') for value in ['نام', 'Name', '/skip', 'خلاصه', '2:28']: self.message(value) self.message(photo=[{'file_id': 'poster', 'file_size': 1}]) self.assertEqual(self.db.one('SELECT * FROM movies')['poster'], 'poster') self.assertEqual(self.db.one('SELECT * FROM movies')['poster_archive'], 0) self.assertEqual(self.db.session(1)['step'], 'trailer') self.assertFalse(any(m == 'copyMessage' for m, _ in self.api.calls)) def test_config_reads_force_subscribe_and_join_url(self): env = {'BOT_TOKEN': '123:' + 'a' * 25, 'ADMIN_IDS': '5', 'CHANNEL_ID': '@testch', 'ARCHIVE_CHANNEL_ID': '-100111', 'FORCE_SUBSCRIBE': 'no', 'JOIN_URL': 'https://t.me/+abc'} saved = {key: os.environ.get(key) for key in env} try: os.environ.update(env) cfg = Config.load() self.assertFalse(cfg.force_sub) self.assertEqual(cfg.join_url, 'https://t.me/+abc') finally: for key, value in saved.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value # ---------- v2.0.0 additions ---------- def test_rating_flow_updates_average_on_card(self): movie = self.draft() self.click(f"publish:{movie['id']}") self.click(f"rate:{movie['id']}", user=2) self.click(f"score:{movie['id']}:8", user=2) self.click(f"rate:{movie['id']}", user=3) self.click(f"score:{movie['id']}:6", user=3) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM user_ratings WHERE movie_id=?', movie['id'])['n'], 2) self.click(f"movie:{movie['id']}", user=4) card = [d for m, d in self.api.calls if m == 'sendMessage'][-1]['text'] self.assertIn('امتیاز کاربران: 7.0', card) self.assertIn('۲ رأی', card) def test_rating_rejected_for_unpublished(self): movie = self.draft() self.click(f"score:{movie['id']}:9", user=2) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM user_ratings')['n'], 0) def test_save_toggle_and_profile_listing(self): movie = self.draft() self.click(f"publish:{movie['id']}") self.click(f"save:{movie['id']}", user=2) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM user_saves')['n'], 1) self.click('saved:0', user=2) self.assertIn('فیلم آزمایشی', str(self.api.calls[-1])) self.click(f"save:{movie['id']}", user=2) self.assertEqual(self.db.one('SELECT COUNT(*) n FROM user_saves')['n'], 0) def test_panel_browse_genre_year_and_top(self): movie = self.draft() self.click(f"publish:{movie['id']}") self.click('genres', user=2) self.assertIn('ژانرها', str(self.api.calls[-1])) self.click('genre:1:0', user=2) self.assertIn('فیلم آزمایشی', str(self.api.calls[-1])) self.click('decades', user=2) self.click('decyear:2020', user=2) self.click('yearpg:2025:0', user=2) self.assertIn('فیلم آزمایشی', str(self.api.calls[-1])) self.click('top:0', user=2) self.assertIn('بهترین امتیازها', str(self.api.calls[-1])) def test_start_menu_and_panel_entry(self): self.message('/start', user=2) self.assertIn('پیشخوان', str(self.api.calls[-1])) self.click('panel', user=2) self.assertIn('پیشخوان کاربری', str(self.api.calls[-1])) self.message('/panel', user=2) self.assertIn('پیشخوان کاربری', str(self.api.calls[-1])) def test_admin_actor_library_and_card(self): self.message('/new') for value in ['نام', 'Name', '/skip', 'خلاصه', '2:28']: self.message(value) self.message(photo=[{'file_id': 'poster'}]) self.message('/skip') mid = self.new_id() self.click(f'catsel:{mid}:1') self.click(f'gendone:{mid}') self.click(f'yrsel:{mid}:2025') self.click(f'actsearch:{mid}') self.message('nothing-found') self.click(f'actnew:{mid}') self.message('Leonardo DiCaprio') self.assertEqual(self.db.one('SELECT COUNT(*) n FROM movie_actors WHERE movie_id=?', mid)['n'], 1) self.click(f'actdone:{mid}') self.message('/skip') self.message('720p') self.message(document={'file_id': 'f', 'file_size': 1}) self.click(f"publish:{mid}") self.click(f"movie:{mid}", user=2) card = [d for m, d in self.api.calls if m == 'sendMessage'][-1]['text'] self.assertIn('Leonardo DiCaprio', card) def test_category_selection_via_editmeta(self): movie = self.draft() self.click(f'editmeta:{movie["id"]}') self.assertEqual(self.db.session(1)['step'], 'category') self.click(f'catsel:{movie["id"]}:2') self.assertEqual(self.db.movie(movie['id'])['category'], 'سریال') # ---------- v2.1.0 additions ---------- def test_duration_parsing_variants(self): self.assertEqual(parse_duration('02:28:03'), '2:28:03') self.assertEqual(parse_duration('0:5:3'), '0:05:03') self.assertEqual(parse_duration('2:28'), '2:28:00') self.assertEqual(parse_duration('۰۲:۲۸:۰۳'), '2:28:03') self.assertEqual(parse_duration('148'), '2:28:00') self.assertEqual(parse_duration('2'), '0:02:00') self.assertIsNone(parse_duration('2:75')) self.assertIsNone(parse_duration('1:2:3:4')) self.assertIsNone(parse_duration('حدود دو ساعت')) def test_duration_display_exact_and_legacy(self): self.assertEqual(duration_display('2:28:03'), '۲:۲۸:۰۳ (۱۴۸ دقیقه)') self.assertEqual(duration_display('2:28:00'), '۲:۲۸:۰۰ (۱۴۸ دقیقه)') self.assertEqual(duration_display('کمتر از ۹۰ دقیقه'), 'کمتر از ۹۰ دقیقه') self.assertEqual(duration_display(''), '') self.assertEqual(duration_display(None), '') def test_duration_typed_step_validation(self): self.message('/new') for value in ['نام', 'Name', '7.5', 'خلاصه']: self.message(value) self.message('نامعتبر') self.assertEqual(self.db.session(1)['step'], 'duration') self.message('۱۴۰ دقیقه') self.assertEqual(self.db.session(1)['step'], 'duration') self.message('02:28:03') self.assertEqual(self.db.one('SELECT duration FROM movies WHERE id=?', self.new_id())['duration'], '2:28:03') self.assertEqual(self.db.session(1)['step'], 'poster') def test_caption_every_item_has_its_own_labeled_line(self): movie = self.draft() text = self.app.movie_card_text(movie) for label in ['نام فیلم:', 'نام لاتین:', 'دسته‌بندی:', 'سال ساخت:', 'امتیاز IMDb:', 'مدت‌زمان:', 'ژانر:', 'کارگردان:', 'خلاصه داستان:']: self.assertIn(label, text) packed = [line for line in text.split('\n') if line.count(' · ') >= 2] self.assertEqual(packed, []) def test_post_caption_shows_user_rating_after_votes(self): movie = self.draft() self.click(f"publish:{movie['id']}") self.click(f"score:{movie['id']}:9", user=2) self.click(f"movie:{movie['id']}", user=3) card = [d for m, d in self.api.calls if m == 'sendMessage'][-1]['text'] self.assertIn('امتیاز کاربران: 9.0 از ۱۰ (۱ رأی)', card) def test_channel_post_caption_live_updates_with_votes(self): movie = self.draft() self.click(f"publish:{movie['id']}") self.click(f"score:{movie['id']}:8", user=2) edits = [d for m, d in self.api.calls if m == 'editMessageCaption' and d.get('chat_id') == '@femonascene'] self.assertEqual(len(edits), 1) self.assertIn('امتیاز کاربران', edits[0]['caption']) self.assertIn('۱ رأی', edits[0]['caption']) self.click(f"score:{movie['id']}:5", user=3) edits = [d for m, d in self.api.calls if m == 'editMessageCaption' and d.get('chat_id') == '@femonascene'] self.assertEqual(len(edits), 1) # throttled: no second edit within the cooldown def test_buttons_carry_telegram_styles(self): movie = self.draft() self.click(f"publish:{movie['id']}") post = next(d for m, d in self.api.calls if m == 'sendPhoto' and d['chat_id'] == '@femonascene') row = post['reply_markup']['inline_keyboard'][0] self.assertEqual(row[0]['style'], 'success') self.assertEqual(row[1]['style'], 'primary') self.click(f"movie:{movie['id']}", user=2) card = [d for m, d in self.api.calls if m == 'sendMessage'][-1]['reply_markup']['inline_keyboard'] self.assertTrue(all(b.get('style') == 'success' for b in card[0])) self.assertTrue(any(b.get('style') == 'danger' for row in card for b in row)) def test_legacy_approx_duration_still_displays(self): movie = self.draft() self.db.run("UPDATE movies SET duration='کمتر از ۹۰ دقیقه' WHERE id=?", movie['id']) text = self.app.movie_card_text(self.db.movie(movie['id'])) self.assertIn('⏱ مدت‌زمان: کمتر از ۹۰ دقیقه', text) # ---------- 3.1.0: config, time-limited links, health endpoint ---------- def variant_of(self, movie): return self.db.one('SELECT * FROM variants WHERE movie_id=?', movie['id']) def test_config_parses_port_link_ttl_and_proxy(self): import unittest.mock old = os.getcwd() os.chdir(self.tmp.name) try: with unittest.mock.patch.dict(os.environ, { 'BOT_TOKEN': '123:tokentokentokentokentokentoken', 'ADMIN_IDS': '1', 'CHANNEL_ID': '@mychannel', 'ARCHIVE_CHANNEL_ID': '-1001234567890', 'PORT': '8000', 'LINK_TTL': '60', 'PROXY_URL': 'http://127.0.0.1:3128'}, clear=True): cfg = Config.load() finally: os.chdir(old) self.assertEqual(cfg.port, 8000) self.assertEqual(cfg.link_ttl, 60) self.assertEqual(cfg.proxy, 'http://127.0.0.1:3128') def test_config_rejects_socks_proxy_and_negative_ttl(self): import unittest.mock old = os.getcwd() os.chdir(self.tmp.name) base = {'BOT_TOKEN': '123:tokentokentokentokentokentoken', 'ADMIN_IDS': '1', 'CHANNEL_ID': '@mychannel', 'ARCHIVE_CHANNEL_ID': '-1001234567890'} try: with unittest.mock.patch.dict(os.environ, {**base, 'PROXY_URL': 'socks5://x:1080'}, clear=True): with self.assertRaises(ValueError): Config.load() with unittest.mock.patch.dict(os.environ, {**base, 'LINK_TTL': '-5'}, clear=True): with self.assertRaises(ValueError): Config.load() with unittest.mock.patch.dict(os.environ, {**base, 'PORT': 'abc'}, clear=True): with self.assertRaises(ValueError): Config.load() finally: os.chdir(old) def test_link_expiry_schedules_and_deletes_after_ttl(self): self.app.cfg.link_ttl = 60 movie = self.draft() self.click(f"publish:{movie['id']}") self.click(f"file:{self.variant_of(movie)['id']}", user=2) self.drain() row = self.db.one('SELECT * FROM expiries') self.assertIsNotNone(row) self.assertEqual(row['chat_id'], 2) # FakeAPI answers message_id = len(self.calls); the delivered file is at index i → id i+1. index = next(i for i, (m, d) in enumerate(self.api.calls) if m in ('sendVideo', 'sendDocument') and d.get('chat_id') == 2) self.assertEqual(row['message_id'], index + 1) self.db.run('UPDATE expiries SET expire_at=?', 1.0) # simulate the passing of the window self.app.expire_due() self.assertEqual([1 for m, d in self.api.calls if m == 'deleteMessage' and d.get('chat_id') == 2], [1]) self.assertIsNone(self.db.one('SELECT * FROM expiries')) notice = [d for m, d in self.api.calls if m == 'sendMessage' and d.get('chat_id') == 2][-1]['text'] self.assertIn('مهلت نمایش این فایل', notice) def test_expiry_api_failure_keeps_job_for_retry(self): self.app.cfg.link_ttl = 60 movie = self.draft() self.click(f"publish:{movie['id']}") self.click(f"file:{self.variant_of(movie)['id']}", user=2) self.drain() self.db.run('UPDATE expiries SET expire_at=?', 1.0) self.api.fail_error = APIError('temporary outage', code=500) self.app.expire_due() row = self.db.one('SELECT * FROM expiries') self.assertIsNotNone(row) self.assertEqual(row['tries'], 1) self.api.fail_error = None self.app.expire_due() self.assertIsNone(self.db.one('SELECT * FROM expiries')) def test_admin_deliveries_never_expire(self): self.app.cfg.link_ttl = 60 movie = self.draft() self.click(f"publish:{movie['id']}") self.click(f"file:{self.variant_of(movie)['id']}", user=1) # uid 1 is an admin self.drain() self.assertIsNone(self.db.one('SELECT * FROM expiries')) self.assertTrue(any(m in ('sendVideo', 'sendDocument') and d.get('chat_id') == 1 for m, d in self.api.calls)) def test_delivery_without_ttl_leaves_no_expiry_rows(self): self.assertEqual(self.app.cfg.link_ttl, 0) # default: feature disabled movie = self.draft() self.click(f"publish:{movie['id']}") self.click(f"file:{self.variant_of(movie)['id']}", user=2) self.drain() self.assertIsNone(self.db.one('SELECT * FROM expiries')) def test_health_endpoint_answers_on_configured_port(self): import json as jsonlib import socket import urllib.request from bot import VERSION as BOT_VERSION, start_health_server probe = socket.socket() probe.bind(('127.0.0.1', 0)) port = probe.getsockname()[1] probe.close() server = start_health_server(port) self.assertIsNotNone(server) try: with urllib.request.urlopen(f'http://127.0.0.1:{port}/', timeout=5) as response: payload = jsonlib.load(response) self.assertEqual(payload['service'], 'FemonaScene') self.assertEqual(payload['version'], BOT_VERSION) self.assertTrue(payload['ok']) finally: server.shutdown() server.server_close() def test_health_server_disabled_when_port_is_zero(self): from bot import start_health_server self.assertIsNone(start_health_server(0)) if __name__ == '__main__': unittest.main()