fc06ec4a12
- Новая страница /scheduled_posts в веб-панели - Создание постов с фото/текстом в любую тему форума - Планирование даты и времени публикации - Автоматическая отправка по расписанию (каждую минуту) - Поддержка всех 10 тем форума - Уведомления админа о публикации/ошибках - Миграция БД: таблица scheduled_posts - Тесты: test_scheduled_posts.py Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
83 lines
3.3 KiB
Python
83 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Тестирование запланированных постов
|
||
"""
|
||
import asyncio
|
||
import aiohttp
|
||
from datetime import datetime, timedelta, timezone
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(BASE_DIR))
|
||
|
||
from config import WEB_ADMIN_LOGIN, WEB_ADMIN_PASSWORD
|
||
|
||
async def test_scheduled_posts():
|
||
"""Тест создание запланированного поста"""
|
||
|
||
# Базовый URL веб-панели
|
||
base_url = "http://localhost:8000"
|
||
|
||
# Данные для аутентификации из конфига
|
||
auth = aiohttp.BasicAuth(WEB_ADMIN_LOGIN, WEB_ADMIN_PASSWORD)
|
||
|
||
print(f"🔐 Используем логин: {WEB_ADMIN_LOGIN}")
|
||
|
||
# Время через 2 минуты от сейчас
|
||
scheduled_time = datetime.now(timezone.utc) + timedelta(minutes=2)
|
||
|
||
# Тест 1: Создание поста
|
||
print("\n📅 Тест 1: Создание запланированного поста...")
|
||
|
||
form_data = aiohttp.FormData()
|
||
form_data.add_field('text', '<b>🗑️ Юмор про мусор #1</b>\n\n<i>Почему мусорный бак всегда полон? Потому что у него нет проблемы с переполнением! 😂</i>')
|
||
form_data.add_field('topic_name', 'memes')
|
||
form_data.add_field('recipients', 'chat_only')
|
||
form_data.add_field('scheduled_time', scheduled_time.isoformat())
|
||
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(
|
||
f"{base_url}/api/scheduled_posts/create",
|
||
auth=auth,
|
||
data=form_data
|
||
) as resp:
|
||
result = await resp.json()
|
||
print(f"Ответ: {result}")
|
||
|
||
if result.get('success'):
|
||
print(f"✅ Пост создан! ID: {result.get('post_id')}")
|
||
else:
|
||
print(f"❌ Ошибка: {result.get('error')}")
|
||
return
|
||
|
||
except Exception as e:
|
||
print(f"❌ Ошибка запроса: {e}")
|
||
return
|
||
|
||
# Тест 2: Получение списка постов
|
||
print("\n📋 Тест 2: Получение списка постов...")
|
||
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(
|
||
f"{base_url}/api/scheduled_posts/list",
|
||
auth=auth
|
||
) as resp:
|
||
result = await resp.json()
|
||
print(f"Найдено постов: {len(result.get('posts', []))}")
|
||
|
||
for post in result.get('posts', []):
|
||
print(f" - ID {post['id']}: {post['topic_name']} в {post['scheduled_time']} [{post['status']}]")
|
||
|
||
except Exception as e:
|
||
print(f"❌ Ошибка запроса: {e}")
|
||
return
|
||
|
||
print("\n✅ Тесты завершены!")
|
||
print(f"⏳ Подожди 2 минуты и проверь что пост опубликовался автоматически!")
|
||
print(f"📱 Тебе придёт уведомление в Telegram когда пост будет опубликован!")
|
||
|
||
if __name__ == '__main__':
|
||
asyncio.run(test_scheduled_posts())
|