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
|
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import threading
import time
import unittest
import deckit.modules.obs as obs
class Fake:
connected = True
def request(self, kind, data=None):
Fake.last = (kind, data)
return {"ok": True}
class Actions(unittest.TestCase):
def setUp(self):
self.m = obs.make({"obs": {"host": "h", "port": 4455, "password": ""}, "client": Fake()})
def test_scene_and_record(self):
self.m.actions["scene"]({"scene": "Live"})
self.assertEqual(Fake.last[0], "SetCurrentProgramScene")
self.m.actions["record"]({"on": True})
self.assertEqual(Fake.last[0], "StartRecord")
self.m.actions["record"]({"on": False})
self.assertEqual(Fake.last[0], "StopRecord")
def test_mute_and_source(self):
self.m.actions["mute"]({"input": "Mic", "muted": True})
self.assertEqual(Fake.last[0], "SetInputMute")
self.m.actions["source"]({"scene": "S", "id": 2, "on": False})
self.assertEqual(Fake.last[0], "SetSceneItemEnabled")
def test_state_offline(self):
class Off:
connected = False
self.assertEqual(obs.make({"client": Off()}).state(), {"connected": False})
class Serialized(unittest.TestCase):
def test_actions_do_not_overlap(self):
active = 0
overlap = []
class Guard:
connected = True
def request(self, kind, data=None):
nonlocal active
active += 1
if active > 1:
overlap.append(kind)
time.sleep(0.02)
active -= 1
return {}
m = obs.make({"client": Guard()})
threads = [threading.Thread(target=m.actions["record"], args=({"on": True},)) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
self.assertEqual(overlap, [])
if __name__ == "__main__":
unittest.main()
|
