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
109
110
111
112
113
114
115
116
117
118
|
from __future__ import print_function, unicode_literals
import json
import pytest
import random
import ssl
import string
import sys
from websocket import ABNF
sys.path.append(".")
import wee_slack
from wee_slack import EventRouter, SlackRequest
class fakewebsocket(object):
def __init__(self):
self.returndata = []
self.sentdata = []
def add(self, data):
self.returndata.append(json.dumps(data).encode('utf-8'))
def recv(self):
return self.recv_data()[1].decode('utf-8')
def recv_data(self, control_frame=False):
if self.returndata:
return ABNF.OPCODE_TEXT, self.returndata.pop(0)
else:
raise ssl.SSLWantReadError()
def send(self, data):
self.sentdata.append(data)
@pytest.fixture
def mock_websocket():
return fakewebsocket()
@pytest.fixture
def realish_eventrouter(mock_websocket, mock_weechat):
e = EventRouter()
wee_slack.EVENTROUTER = e
context = e.store_context(SlackRequest('xoxs-token', 'rtm.start', {}))
with open('_pytest/data/http/rtm.start.json') as rtmstartfile:
if sys.version_info.major == 2:
rtmstartdata = rtmstartfile.read().decode('utf-8')
else:
rtmstartdata = rtmstartfile.read()
e.receive_httprequest_callback(context, '', 0, rtmstartdata, '')
while len(e.queue):
e.handle_next()
for team in e.teams.values():
team.ws = mock_websocket
return e
@pytest.fixture
def team(realish_eventrouter):
return next(iter(realish_eventrouter.teams.values()))
@pytest.fixture
def channel_general(team):
return team.channels[team.get_channel_map()['general']]
@pytest.fixture
def user_alice(team):
return team.users[team.get_username_map()['alice']]
class FakeWeechat():
"""
this is the thing that acts as "w." everywhere..
basically mock out all of the weechat calls here i guess
"""
WEECHAT_RC_ERROR = 0
WEECHAT_RC_OK = 1
WEECHAT_RC_OK_EAT = 2
def __init__(self):
self.config = {}
def prnt(*args):
output = "("
for arg in args:
if arg != None:
output += "{}, ".format(arg)
print("w.prnt {}".format(output))
def hdata_get(*args):
return "0x000001"
def hdata_pointer(*args):
return "0x000002"
def hdata_time(*args):
return "1355517519"
def hdata_string(*args):
return "testuser"
def buffer_new(*args):
return "0x" + "".join(random.choice(string.digits) for _ in range(8))
def prefix(self, type):
return ""
def config_get_plugin(self, key):
return self.config.get(key, "")
def config_get(self, key):
return ""
def config_set_plugin(self, key, value):
self.config[key] = value
def config_string(self, key):
return ""
def color(self, name):
return ""
def __getattr__(self, name):
def method(*args):
pass
return method
@pytest.fixture
def mock_weechat():
wee_slack.w = FakeWeechat()
wee_slack.config = wee_slack.PluginConfig()
wee_slack.hdata = wee_slack.Hdata(wee_slack.w)
wee_slack.debug_string = None
wee_slack.slack_debug = "debug_buffer_ptr"
wee_slack.STOP_TALKING_TO_SLACK = False
wee_slack.proc = {}
wee_slack.weechat_version = 0x10500000
|
