From 16ee57b1581fd63cefbc7bca44493760ab028ed4 Mon Sep 17 00:00:00 2001 From: DarkShyMW Date: Sat, 23 Aug 2025 09:54:40 +0300 Subject: [PATCH] Add tests --- .github/workflows/tests.yml | 29 +++++++++++ README.md | 1 + app.py | 95 ++++++++++++++++++++++++------------- requirements.txt | 6 ++- tests/test_fetch.py | 51 ++++++++++++++++++++ tests/test_post.py | 24 ++++++++++ 6 files changed, 170 insertions(+), 36 deletions(-) create mode 100644 .github/workflows/tests.yml create mode 100644 tests/test_fetch.py create mode 100644 tests/test_post.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..4f1a4e0 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,29 @@ +name: Tests + +on: + push: + branches: [ "main" ] + pull_request: + branches: [ "main" ] + +jobs: + test: + runs-on: ubuntu-latest + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + pip install pytest pytest-asyncio + + - name: Run tests + run: pytest diff --git a/README.md b/README.md index dd6d138..e11b18c 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,5 @@ # 🐴 Derpibooru Telegram Bot +![Tests](https://github.com/DarkShyMW/derpibooru-telegram-bot/actions/workflows/tests.yml/badge.svg) Telegram-Π±ΠΎΡ‚, ΠΊΠΎΡ‚ΠΎΡ€Ρ‹ΠΉ автоматичСски ΠΏΡƒΠ±Π»ΠΈΠΊΡƒΠ΅Ρ‚ изобраТСния с [Derpibooru](https://derpibooru.org) ΠΏΠΎ Π·Π°Π΄Π°Π½Π½Ρ‹ΠΌ Ρ‚Π΅Π³Π°ΠΌ. diff --git a/app.py b/app.py index ed08f78..43295a8 100644 --- a/app.py +++ b/app.py @@ -3,26 +3,28 @@ import aiohttp import logging import random import json -from datetime import datetime, timedelta -from telegram import Bot -from pathlib import Path import sys +from datetime import datetime, timedelta +from pathlib import Path +from telegram import Bot # === Настройки === -TELEGRAM_TOKEN = "************************" -CHANNEL_ID = -100*********** # числовой ID ΠΊΠ°Π½Π°Π»Π° -DERPIBOORU_TOKEN = "Ρ‚ΠΎΠΊΠ΅Π½ ΠΎΡ‚ Π΄Π΅Ρ€ΠΏΠΈΠ±ΡƒΡ€Ρƒ" +TELEGRAM_TOKEN = "Ρ‚ΠΎΠΊΠ΅Π½_вашСго_Ρ‚Π³_Π°ΠΊΠΊΠ°ΡƒΠ½Ρ‚Π°" # Π’ΠΎΠΊΠ΅Π½ ΠΎΡ‚ Telegram +CHANNEL_ID = -100000000000000 # числовой ID ΠΊΠ°Π½Π°Π»Π° +DERPIBOORU_TOKEN = "xxxxxxxxxxxxxxxx" # Π’ΠΎΠΊΠ΅Π½ ΠΎΡ‚ DerpiBooru + TAGS_LIST = [ - ["Ρ‚Π΅Π³1"], - ["Ρ‚Π΅Π³2"], - ["Ρ‚Π΅Π³3"], - ["Ρ‚Π΅Π³4"] + ["gay"], + ["female"], + ["vulva"], + ["creampie"], ] -DERPYBOORU_API_SEARCH = "https://derpibooru.org/api/v1/json/search/images" + +DERPIBOORU_API = "https://derpibooru.org/api/v1/json/search/images" SENT_IMAGES_FILE = Path("sent_images.json") FILTER_ID = 56027 # ID Ρ„ΠΈΠ»ΡŒΡ‚Ρ€Π°, Ρ€Π°Π·Ρ€Π΅ΡˆΠ°ΡŽΡ‰Π΅Π³ΠΎ NSFW изобраТСния -# Π›ΠΎΠ³ΠΈ +# === Π›ΠΎΠ³ΠΈ === logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", @@ -34,7 +36,7 @@ logging.basicConfig( bot = Bot(token=TELEGRAM_TOKEN) -# Π—Π°Π³Ρ€ΡƒΠ·ΠΊΠ° ΡƒΠΆΠ΅ ΠΎΡ‚ΠΏΡ€Π°Π²Π»Π΅Π½Π½Ρ‹Ρ… ΠΈΠ·ΠΎΠ±Ρ€Π°ΠΆΠ΅Π½ΠΈΠΉ +# === Π—Π°Π³Ρ€ΡƒΠΆΠ°Π΅ΠΌ ΡƒΠΆΠ΅ ΠΎΡ‚ΠΏΡ€Π°Π²Π»Π΅Π½Π½Ρ‹Π΅ изобраТСния === if SENT_IMAGES_FILE.exists(): try: with SENT_IMAGES_FILE.open("r", encoding="utf-8") as f: @@ -47,11 +49,12 @@ else: async def fetch_image(session, tags): + """Запрос изобраТСния с Derpibooru ΠΏΠΎ Ρ‚Π΅Π³Π°ΠΌ.""" params = { "q": " ".join(tags), "per_page": 50, "page": 1, - "key": DERPIBOORU_TOKEN + "key": DERPIBOORU_TOKEN, } if FILTER_ID: params["filter_id"] = FILTER_ID @@ -59,10 +62,11 @@ async def fetch_image(session, tags): backoff = 1 while True: try: - async with session.get(DERPYBOORU_API_SEARCH, params=params) as response: + async with session.get(DERPIBOORU_API, params=params) as response: if response.status == 200: data = await response.json() images = data.get("images", []) + if not images: logging.warning("НичСго Π½Π΅ Π½Π°ΠΉΠ΄Π΅Π½ΠΎ ΠΏΠΎ Ρ‚Π΅Π³Π°ΠΌ: %s", tags) return None @@ -71,6 +75,7 @@ async def fetch_image(session, tags): for img in images: reps = img.get("representations", {}) url = reps.get("full") or reps.get("large") or reps.get("medium") + if not url: continue if url not in sent_images: @@ -80,16 +85,21 @@ async def fetch_image(session, tags): logging.info("Π’Ρ‹Π±Ρ€Π°Π½ΠΎ ΠΈΠ·ΠΎΠ±Ρ€Π°ΠΆΠ΅Π½ΠΈΠ΅: %s", url) return url, author, source, img_tags - logging.info("ВсС Π½Π°ΠΉΠ΄Π΅Π½Π½Ρ‹Π΅ изобраТСния ΡƒΠΆΠ΅ ΠΎΡ‚ΠΏΡ€Π°Π²Π»Π΅Π½Ρ‹ для Ρ‚Π΅Π³ΠΎΠ²: %s", tags) + logging.info("ВсС Π½Π°ΠΉΠ΄Π΅Π½Π½Ρ‹Π΅ изобраТСния ΡƒΠΆΠ΅ ΠΎΡ‚ΠΏΡ€Π°Π²Π»Π΅Π½Ρ‹: %s", tags) return None - elif response.status in [500, 501]: - logging.warning("Ошибка %s ΠΎΡ‚ Derpibooru. Π–Π΄Π΅ΠΌ %s сСкунд ΠΏΠ΅Ρ€Π΅Π΄ ΠΏΠΎΠ²Ρ‚ΠΎΡ€ΠΎΠΌ...", response.status, backoff) + elif response.status in (500, 501): + logging.warning( + "Ошибка %s ΠΎΡ‚ Derpibooru. Π–Π΄Π΅ΠΌ %s сСкунд...", + response.status, backoff + ) await asyncio.sleep(backoff) backoff = min(backoff * 2, 900) + else: logging.error("Ошибка API Derpibooru: %s", response.status) return None + except aiohttp.ClientError as e: logging.error("Ошибка запроса ΠΊ Derpibooru: %s", e) await asyncio.sleep(backoff) @@ -97,19 +107,31 @@ async def fetch_image(session, tags): async def post_image(tags=None): + """ΠŸΡƒΠ±Π»ΠΈΠΊΠ°Ρ†ΠΈΡ изобраТСния Π² ΠΊΠ°Π½Π°Π».""" async with aiohttp.ClientSession() as session: if tags is None: tags = random.choice(TAGS_LIST) + result = await fetch_image(session, tags) if result: url, author, source, img_tags = result - caption = f"Автор: {author}\nΠ˜ΡΡ‚ΠΎΡ‡Π½ΠΈΠΊ: {source}\nΠ’Π΅Π³ΠΈ: {', '.join(img_tags)}" + caption = ( + f"Автор: {author}\n" + f"Π˜ΡΡ‚ΠΎΡ‡Π½ΠΈΠΊ: {source}\n" + f"Π’Π΅Π³ΠΈ: {', '.join(img_tags)}" + ) try: - await bot.send_photo(chat_id=CHANNEL_ID, photo=url, caption=caption) + await bot.send_photo( + chat_id=CHANNEL_ID, + photo=url, + caption=caption + ) logging.info("ΠžΡ‚ΠΏΡ€Π°Π²Π»Π΅Π½ΠΎ ΠΈΠ·ΠΎΠ±Ρ€Π°ΠΆΠ΅Π½ΠΈΠ΅: %s", url) + sent_images.add(url) with SENT_IMAGES_FILE.open("w", encoding="utf-8") as f: json.dump(list(sent_images), f) + except Exception as e: logging.error("Ошибка ΠΏΡ€ΠΈ ΠΎΡ‚ΠΏΡ€Π°Π²ΠΊΠ΅ изобраТСния: %s", e) else: @@ -117,15 +139,19 @@ async def post_image(tags=None): async def scheduler(): + """ΠŸΡƒΠ±Π»ΠΈΠΊΡƒΠ΅Ρ‚ ΠΈΠ·ΠΎΠ±Ρ€Π°ΠΆΠ΅Π½ΠΈΠ΅ ΠΊΠ°ΠΆΠ΄Ρ‹ΠΉ час (Π² Π½Π°Ρ‡Π°Π»Π΅ часа).""" while True: now = datetime.now() - # ΠŸΡƒΠ±Π»ΠΈΠΊΡƒΠ΅ΠΌ Π΄ΠΎΠ³ΠΎΠ½ΡΡŽΡ‰Π΅Π΅ ΠΈΠ·ΠΎΠ±Ρ€Π°ΠΆΠ΅Π½ΠΈΠ΅, Ссли запуск Π½Π΅ Π² Π½Π°Ρ‡Π°Π»Π΅ часа + + # Если Π±ΠΎΡ‚ Π·Π°ΠΏΡƒΡ‰Π΅Π½ Π½Π΅ Π² Π½Π°Ρ‡Π°Π»Π΅ часа β€” ΠΏΡƒΠ±Π»ΠΈΠΊΡƒΠ΅ΠΌ сразу if now.minute != 0 or now.second != 0: logging.info("ΠŸΡ€ΠΎΠΏΡƒΡ‰Π΅Π½Π½Ρ‹ΠΉ час β€” ΠΏΡƒΠ±Π»ΠΈΠΊΡƒΠ΅ΠΌ сразу") await post_image() - # Π–Π΄Π΅ΠΌ Π΄ΠΎ ΡΠ»Π΅Π΄ΡƒΡŽΡ‰Π΅ΠΉ 00 ΠΌΠΈΠ½ΡƒΡ‚Ρ‹ - next_run = (now + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0) + # Π–Π΄Π΅ΠΌ Π΄ΠΎ ΡΠ»Π΅Π΄ΡƒΡŽΡ‰Π΅Π³ΠΎ часа + next_run = (now + timedelta(hours=1)).replace( + minute=0, second=0, microsecond=0 + ) wait_seconds = (next_run - datetime.now()).total_seconds() logging.info("Π–Π΄Π΅ΠΌ %.0f сСкунд Π΄ΠΎ ΡΠ»Π΅Π΄ΡƒΡŽΡ‰Π΅ΠΉ ΠΏΡƒΠ±Π»ΠΈΠΊΠ°Ρ†ΠΈΠΈ", wait_seconds) await asyncio.sleep(wait_seconds) @@ -135,34 +161,35 @@ async def scheduler(): async def command_listener(): + """Π‘Π»ΡƒΡˆΠ°Π΅Ρ‚ ΠΊΠΎΠΌΠ°Π½Π΄Ρ‹ ΠΈΠ· stdin (postnow, posttags, timetopost).""" loop = asyncio.get_event_loop() while True: cmd = await loop.run_in_executor(None, sys.stdin.readline) cmd = cmd.strip().lower() - if cmd == "postnow" or cmd == ">> postnow": - logging.info("Команда postnow ΠΏΠΎΠ»ΡƒΡ‡Π΅Π½Π°, ΠΏΡƒΠ±Π»ΠΈΠΊΡƒΠ΅ΠΌ случайноС ΠΈΠ·ΠΎΠ±Ρ€Π°ΠΆΠ΅Π½ΠΈΠ΅...") + + if cmd in {"postnow", ">> postnow"}: + logging.info("Команда postnow ΠΏΠΎΠ»ΡƒΡ‡Π΅Π½Π° β€” ΠΏΡƒΠ±Π»ΠΈΠΊΡƒΠ΅ΠΌ...") await post_image() + elif cmd.startswith("posttags "): tags = cmd[len("posttags "):].split() - logging.info("Команда posttags ΠΏΠΎΠ»ΡƒΡ‡Π΅Π½Π°, ΠΏΡƒΠ±Π»ΠΈΠΊΡƒΠ΅ΠΌ ΠΈΠ·ΠΎΠ±Ρ€Π°ΠΆΠ΅Π½ΠΈΠ΅ с Ρ‚Π΅Π³Π°ΠΌΠΈ: %s", tags) + logging.info("Команда posttags ΠΏΠΎΠ»ΡƒΡ‡Π΅Π½Π°: %s", tags) await post_image(tags) + elif cmd.startswith("timetopost"): now = datetime.now() - # ΠŸΡƒΠ±Π»ΠΈΠΊΡƒΠ΅ΠΌ Π΄ΠΎΠ³ΠΎΠ½ΡΡŽΡ‰Π΅Π΅ ΠΈΠ·ΠΎΠ±Ρ€Π°ΠΆΠ΅Π½ΠΈΠ΅, Ссли запуск Π½Π΅ Π² Π½Π°Ρ‡Π°Π»Π΅ часа - if now.minute != 0 or now.second != 0: - logging.info("ΠŸΡ€ΠΎΠΏΡƒΡ‰Π΅Π½Π½Ρ‹ΠΉ час β€” ΠΏΡƒΠ±Π»ΠΈΠΊΡƒΠ΅ΠΌ сразу") - await post_image() - - # Π–Π΄Π΅ΠΌ Π΄ΠΎ ΡΠ»Π΅Π΄ΡƒΡŽΡ‰Π΅ΠΉ 00 ΠΌΠΈΠ½ΡƒΡ‚Ρ‹ - next_run = (now + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0) + next_run = (now + timedelta(hours=1)).replace( + minute=0, second=0, microsecond=0 + ) wait_seconds = (next_run - datetime.now()).total_seconds() logging.info("Π’Ρ€Π΅ΠΌΠ΅Π½ΠΈ Π΄ΠΎ ΠΏΡƒΠ±Π»ΠΈΠΊΠ°Ρ†ΠΈΠΈ: %s", wait_seconds) - + else: logging.info("НСизвСстная ΠΊΠΎΠΌΠ°Π½Π΄Π°: %s", cmd) async def main(): + """Π“Π»Π°Π²Π½Ρ‹ΠΉ запуск: ΠΏΠ»Π°Π½ΠΈΡ€ΠΎΠ²Ρ‰ΠΈΠΊ + ΠΊΠΎΠΌΠ°Π½Π΄Ρ‹.""" await asyncio.gather( scheduler(), command_listener() diff --git a/requirements.txt b/requirements.txt index 3b18281..e4eb679 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,4 @@ -python-telegram-bot==20.5 -aiohttp==3.9.5 +pytest +pytest-asyncio +aiohttp +aioresponses diff --git a/tests/test_fetch.py b/tests/test_fetch.py new file mode 100644 index 0000000..2311b83 --- /dev/null +++ b/tests/test_fetch.py @@ -0,0 +1,51 @@ +import pytest +from aioresponses import aioresponses +from app import fetch_image, DERPYBOORU_API_SEARCH + + +@pytest.mark.asyncio +async def test_fetch_image_success(): + """ВСстируСм ΡƒΡΠΏΠ΅ΡˆΠ½Ρ‹ΠΉ Π²ΠΎΠ·Π²Ρ€Π°Ρ‚ ΠΊΠ°Ρ€Ρ‚ΠΈΠ½ΠΊΠΈ ΠΈΠ· API""" + tags = ["gay"] + + mock_data = { + "images": [ + { + "id": 123, + "uploader": "tester", + "view_url": "https://derpibooru.org/images/123", + "tags": ["gay", "cute"], + "representations": { + "full": "https://example.com/full.png", + "large": "https://example.com/large.png" + } + } + ] + } + + with aioresponses() as mocked: + mocked.get(DERPYBOORU_API_SEARCH, status=200, payload=mock_data) + + async with __import__("aiohttp").ClientSession() as session: + result = await fetch_image(session, tags) + + assert result is not None + url, author, source, img_tags = result + assert url == "https://example.com/full.png" + assert author == "tester" + assert source == "https://derpibooru.org/images/123" + assert "gay" in img_tags + + +@pytest.mark.asyncio +async def test_fetch_image_empty(): + """ВСстируСм ΡΠΈΡ‚ΡƒΠ°Ρ†ΠΈΡŽ, ΠΊΠΎΠ³Π΄Π° API Π½Π΅ Π²Π΅Ρ€Π½ΡƒΠ» ΠΊΠ°Ρ€Ρ‚ΠΈΠ½ΠΎΠΊ""" + tags = ["nonexistent"] + + with aioresponses() as mocked: + mocked.get(DERPYBOORU_API_SEARCH, status=200, payload={"images": []}) + + async with __import__("aiohttp").ClientSession() as session: + result = await fetch_image(session, tags) + + assert result is None diff --git a/tests/test_post.py b/tests/test_post.py new file mode 100644 index 0000000..9547133 --- /dev/null +++ b/tests/test_post.py @@ -0,0 +1,24 @@ +import pytest +from unittest.mock import AsyncMock, patch +from app import post_image + + +@pytest.mark.asyncio +@patch("app.bot.send_photo", new_callable=AsyncMock) +@patch("app.fetch_image", return_value=( + "https://example.com/full.png", + "tester", + "https://derpibooru.org/images/123", + ["gay", "cute"] +)) +async def test_post_image_success(mock_fetch, mock_send_photo, tmp_path, monkeypatch): + """ВСстируСм ΡƒΡΠΏΠ΅ΡˆΠ½ΡƒΡŽ ΠΏΡƒΠ±Π»ΠΈΠΊΠ°Ρ†ΠΈΡŽ ΠΊΠ°Ρ€Ρ‚ΠΈΠ½ΠΊΠΈ""" + + # Π²Ρ€Π΅ΠΌΠ΅Π½Π½Ρ‹ΠΉ Ρ„Π°ΠΉΠ» для sent_images + monkeypatch.setattr("app.SENT_IMAGES_FILE", tmp_path / "sent.json") + monkeypatch.setattr("app.sent_images", set()) + + await post_image(tags=["gay"]) + + mock_fetch.assert_called_once() + mock_send_photo.assert_awaited_once()