Spaces:
Running
Running
File size: 3,762 Bytes
5f400f8 21cde0c 5f400f8 21cde0c 5f400f8 6c3a31d b737d7a 6c3a31d b737d7a 6c3a31d d55a0af b737d7a d55a0af 5f400f8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """Shared fixture and helpers for the handler-level tests (real Telethon event builders, fake events and client)."""
import io
from types import SimpleNamespace
import pytest
from PIL import Image
from bot.app import App
from bot.config import load_config
from bot.db import Database
from .tgfakes import FakeCallback, FakeClient, FakeEvent, dispatch, document_message, flat_buttons, telegram_text
OWNER, ADMIN_CFG, ADMIN_DB, STRANGER = 1000001, 1000002, 1000003, 1999999
LOG = -1001111111111
EPISODE = "[SubsPlease] Tokyo Revengers - 05 (1080p) [ABCD1234].mkv"
EPISODE_NEW = "[S01E05] Tokyo Revengers [1080p].mkv"
def env_for(tmp_path, **extra):
return {
"API_ID": "1", "API_HASH": "h", "BOT_TOKEN": "1:t", "OWNER_ID": str(OWNER), "ADMIN_IDS": str(ADMIN_CFG),
"LOG_CHANNEL": str(LOG), "TEMP_DIR": str(tmp_path / "work"), "FAST_TRANSFER": "0", "MAX_FILE_MB": "100",
**extra,
}
@pytest.fixture
async def ctx(tmp_path):
cfg = load_config(env_for(tmp_path))
db = Database(f"sqlite+aiosqlite:///{tmp_path}/app.db")
await db.init()
client = FakeClient()
app = App(client, cfg, db)
app._register() # setup() minus the queue workers: jobs stay queued so they can be inspected
yield SimpleNamespace(app=app, client=client, db=db, cfg=cfg, tmp=tmp_path)
await app.shutdown()
await db.close()
async def say(ctx, sender, text="", **kw):
ev = FakeEvent(sender, text, **kw)
await dispatch(ctx.client, ev)
return ev
async def say_formatted(ctx, sender, markup, **kw):
"""Send a message with formatting, written like ``"[b]bold[/b] [i]italic[/i] [a=https://x.io]link[/a]"`` (see ``telegram_text``)."""
text, entities = telegram_text(markup)
return await say(ctx, sender, text, entities=entities, **kw)
async def send_doc(ctx, sender, name=EPISODE, size=50 * 1024 * 1024, **kw):
message = document_message(name, size, **kw)
ctx.client.messages[message.id] = message # so get_messages() can find it again
ev = FakeEvent(sender, message=message)
await dispatch(ctx.client, ev)
return ev
async def click(ctx, sender, data, **kw):
cb = FakeCallback(sender, data, **kw)
await dispatch(ctx.client, cb)
return cb
async def prefs(ctx, uid=OWNER):
return await ctx.db.get_settings(uid)
async def autofmt(ctx, uid=OWNER):
"""The auto format that renames the user's files (``None`` when they have none)."""
return await ctx.db.get_auto_format(uid)
async def auto_thumb(ctx, uid=OWNER):
"""The thumbnail of the user's auto format (``None`` when there is none)."""
fmt = await autofmt(ctx, uid)
return await ctx.db.get_format_thumb(uid, fmt["id"]) if fmt is not None and fmt["has_thumb"] else None
async def ensure_auto(db, uid=OWNER):
"""The user's auto format, made first (a format called "Default") when they have none: what the tests build on."""
return await db.get_auto_format(uid) or await db.create_format(uid, "Default", kind="auto")
async def set_auto_thumb(ctx, data, uid=OWNER):
"""Give the user's auto format a thumbnail, making the "Default" auto format first if they have none."""
fmt = await ensure_auto(ctx.db, uid)
await ctx.db.set_format_thumb(uid, fmt["id"], data)
return fmt
async def edit_auto(ctx, uid=OWNER, **changes):
"""Change the user's auto format (``template=...``, ``caption=...``), making the "Default" one first if they have none."""
fmt = await ensure_auto(ctx.db, uid)
return await ctx.db.update_format(uid, fmt["id"], **changes)
def jpeg(size=(640, 480)):
buf = io.BytesIO()
Image.new("RGB", size, (200, 30, 30)).save(buf, "JPEG")
return buf.getvalue()
def data_of(buttons):
return [d for _l, d in flat_buttons(buttons)]
|