136 lines
3.7 KiB
Python
136 lines
3.7 KiB
Python
import pytest
|
|
import json
|
|
|
|
from utils import *
|
|
from contextlib import contextmanager
|
|
from fbchat import Message, Mention
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def session():
|
|
class FakeSession:
|
|
# TODO: Add a further mocked session
|
|
user_id = "31415926536"
|
|
|
|
def __repr__(self):
|
|
return "<FakeSession>"
|
|
|
|
return FakeSession()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def user(client2):
|
|
return {"id": client2.id, "type": None}
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def group(pytestconfig):
|
|
return {
|
|
"id": load_variable("group_id", pytestconfig.cache),
|
|
"type": None,
|
|
}
|
|
|
|
|
|
@pytest.fixture(
|
|
scope="session",
|
|
params=["user", "group", pytest.param("none", marks=[pytest.mark.xfail()])],
|
|
)
|
|
def thread(request, user, group):
|
|
return {"user": user, "group": group, "none": {"id": "0", "type": None},}[
|
|
request.param
|
|
]
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def client1(pytestconfig):
|
|
with load_client(1, pytestconfig.cache) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def client2(pytestconfig):
|
|
with load_client(2, pytestconfig.cache) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client(client1, thread):
|
|
# TODO: These are commented out since `setDefaultThread` is removed - But now any
|
|
# test that use this fixture is likely broken!
|
|
# client1.setDefaultThread(thread["id"], thread["type"])
|
|
yield client1
|
|
# client1.resetDefaultThread()
|
|
|
|
|
|
@pytest.fixture(scope="session", params=["client1", "client2"])
|
|
def client_all(request, client1, client2):
|
|
return client1 if request.param == "client1" else client2
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def catch_event(client2):
|
|
t = ClientThread(client2)
|
|
t.start()
|
|
|
|
@contextmanager
|
|
def inner(method_name):
|
|
caught = CaughtValue()
|
|
old_method = getattr(client2, method_name)
|
|
|
|
# Will be called by the other thread
|
|
def catch_value(*args, **kwargs):
|
|
old_method(*args, **kwargs)
|
|
# Make sure the `set` is only called once
|
|
if not caught.is_set():
|
|
caught.set(kwargs)
|
|
|
|
setattr(client2, method_name, catch_value)
|
|
yield caught
|
|
caught.wait()
|
|
if not caught.is_set():
|
|
raise ValueError("The value could not be caught")
|
|
setattr(client2, method_name, old_method)
|
|
|
|
yield inner
|
|
|
|
t.should_stop.set()
|
|
|
|
try:
|
|
# Make the client send a messages to itself, so the blocking pull request will return
|
|
# This is probably not safe, since the client is making two requests simultaneously
|
|
client2.send(Message(text=random_hex()), client2.id)
|
|
finally:
|
|
t.join()
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def compare(client, thread):
|
|
def inner(caught_event, **kwargs):
|
|
d = {
|
|
"author_id": client.id,
|
|
"thread_id": client.id if thread["type"] == None else thread["id"],
|
|
"thread_type": thread["type"],
|
|
}
|
|
d.update(kwargs)
|
|
return subset(caught_event.res, **d)
|
|
|
|
return inner
|
|
|
|
|
|
@pytest.fixture(params=["me", "other", "me other"])
|
|
def message_with_mentions(request, client, client2, group):
|
|
text = "Hi there ["
|
|
mentions = []
|
|
if "me" in request.param:
|
|
mentions.append(Mention(thread_id=client.id, offset=len(text), length=2))
|
|
text += "me, "
|
|
if "other" in request.param:
|
|
mentions.append(Mention(thread_id=client2.id, offset=len(text), length=5))
|
|
text += "other, "
|
|
# Unused, because Facebook don't properly support sending mentions with groups as targets
|
|
if "group" in request.param:
|
|
mentions.append(Mention(thread_id=group["id"], offset=len(text), length=5))
|
|
text += "group, "
|
|
text += "nothing]"
|
|
return Message(text, mentions=mentions)
|