-
-
Notifications
You must be signed in to change notification settings - Fork 1
/
conftest.py
66 lines (54 loc) · 1.67 KB
/
conftest.py
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
import asyncio
import os
import sys
import time
from threading import Thread
import pytest
from pyrogram import Client
from pyrogram.types import Chat
from tgintegration import BotController
from fa_search_bot.bot import FASearchBot
@pytest.fixture(scope="session", autouse=True)
def event_loop(request):
""" Create an instance of the default event loop for the session. """
if sys.platform == 'win32':
loop = asyncio.ProactorEventLoop()
asyncio.set_event_loop(loop)
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="session")
async def client() -> Client:
client = Client(
os.getenv("CLIENT_SESSION_STRING"),
api_id=os.getenv("CLIENT_API_ID"),
api_hash=os.getenv("CLIENT_API_HASH")
)
await client.start()
yield client
await client.stop()
@pytest.fixture(scope="session")
def bot() -> FASearchBot:
bot = FASearchBot(os.getenv('CONFIG_FILE', 'config.json'))
if os.getenv("BOT_KEY"):
bot.bot_key = os.getenv("BOT_KEY")
bot_thread = Thread(target=bot.start)
bot_thread.start()
while not bot.alive:
time.sleep(0.1)
yield bot
bot.alive = False
bot_thread.join()
@pytest.fixture
async def controller(client, bot) -> BotController:
controller = BotController(
peer=bot.bot.username,
client=client
)
await controller.initialize(start_client=False)
return controller
@pytest.fixture(scope="session")
async def group_chat(client, bot) -> Chat:
group = await client.create_group("Automated test group", [bot.bot.username])
yield group
await client.leave_chat(group.id, delete=True)