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
|
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import json
import tempfile
import threading
import time
import unittest
from pathlib import Path
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
st = obs.make({"client": Off()}).state()
self.assertFalse(st["connected"])
self.assertIn("error", st)
def test_record_requires_a_bool(self):
for bad in ({}, {"on": "false"}, {"on": 1}):
with self.assertRaises(obs.ModuleError, msg=bad):
self.m.actions["record"](bad)
class Password(unittest.TestCase):
def test_reads_the_password_obs_stores(self):
p = Path(tempfile.mkdtemp()) / "config.json"
p.write_text(json.dumps({"server_password": "sekret"}))
self.assertEqual(obs._obs_password(p), "sekret")
def test_missing_config_is_empty(self):
self.assertEqual(obs._obs_password(Path("/nonexistent/x.json")), "")
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()
|
