-
Notifications
You must be signed in to change notification settings - Fork 60
/
Copy pathtest_ci.py
118 lines (91 loc) · 3.91 KB
/
test_ci.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
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 unittest.mock import patch, Mock
from queue import Queue, Empty
import time
from obswebsocket import obsws, requests, events
def fake_send(data):
if data == '{"request-type": "GetAuthRequired", "message-id": "1"}':
fake_recv.queue.put('{"status":"ok"}')
elif data == '{"message-id": "2", "request-type": "FakeRequest"}':
fake_recv.queue.put('{"message-id": "2", "status":"ok"}')
elif data == '{"op": 1, "d": {"rpcVersion": 1, "authentication": "", "eventSubscriptions": 1023}}':
fake_recv.queue.put('{"op": 2, "d": {"negotiatedRpcVersion": 1 }}')
elif data == '{"op": 6, "d": {"requestId": "1", "requestType": "FakeRequest", "requestData": {}}}':
fake_recv.queue.put('{"op": 7, "d": {"requestId": "1", "requestType": "FakeRequest", "requestStatus": {"result": true, "code": 100}}}')
else:
raise Exception(data)
def fake_recv():
try:
return fake_recv.queue.get(timeout=1)
except Empty:
return ""
fake_recv.queue = Queue() # noqa: E305
def test_request_legacy():
ws = obsws("127.0.0.1", 4444, "", legacy=True)
with patch('websocket.WebSocket') as mock:
mockws = mock.return_value
mockws.send = Mock(wraps=fake_send)
mockws.recv = Mock(wraps=fake_recv)
ws.connect()
mockws.connect.assert_called_once_with("ws://127.0.0.1:4444")
assert ws.thread_recv.running
r = ws.call(requests.FakeRequest())
assert r.name == "FakeRequest"
assert r.status
ws.disconnect()
assert not ws.thread_recv
def test_event_legacy():
ws = obsws("127.0.0.1", 4444, "", legacy=True)
with patch('websocket.WebSocket') as mock:
mockws = mock.return_value
mockws.send = Mock(wraps=fake_send)
mockws.recv = Mock(wraps=fake_recv)
ws.connect()
mockws.connect.assert_called_once_with("ws://127.0.0.1:4444")
assert ws.thread_recv.running
def on_fake_event(message):
assert message.name == "FakeEvent"
assert message.getFakeKey() == "fakeValue"
on_fake_event.ok = True
on_fake_event.ok = False
ws.register(on_fake_event, events.FakeEvent)
fake_recv.queue.put('{"update-type": "FakeEvent", "fakeKey":"fakeValue"}')
time.sleep(1)
assert on_fake_event.ok
ws.disconnect()
assert not ws.thread_recv
def test_request():
ws = obsws("127.0.0.1", 4455, "")
with patch('websocket.WebSocket') as mock:
mockws = mock.return_value
mockws.send = Mock(wraps=fake_send)
mockws.recv = Mock(wraps=fake_recv)
fake_recv.queue.put('{"op": 0, "d": { "obsWebSocketVersion": "fake", "rpcVersion": 1 }}')
ws.connect()
mockws.connect.assert_called_once_with("ws://127.0.0.1:4455")
assert ws.thread_recv.running
r = ws.call(requests.FakeRequest())
assert r.name == "FakeRequest"
assert r.status
ws.disconnect()
assert not ws.thread_recv
def test_event():
ws = obsws("127.0.0.1", 4455, "")
with patch('websocket.WebSocket') as mock:
mockws = mock.return_value
mockws.send = Mock(wraps=fake_send)
mockws.recv = Mock(wraps=fake_recv)
fake_recv.queue.put('{"op": 0, "d": { "obsWebSocketVersion": "fake", "rpcVersion": 1 }}')
ws.connect()
mockws.connect.assert_called_once_with("ws://127.0.0.1:4455")
assert ws.thread_recv.running
def on_fake_event(message):
assert message.name == "FakeEvent"
assert message.getFakeKey() == "fakeValue"
on_fake_event.ok = True
on_fake_event.ok = False
ws.register(on_fake_event, events.FakeEvent)
fake_recv.queue.put('{"op": 5, "d": { "eventType": "FakeEvent", "eventIntent": 1, "eventData": { "fakeKey": "fakeValue" }}}')
time.sleep(1)
assert on_fake_event.ok
ws.disconnect()
assert not ws.thread_recv