Repository navigation
Expand file tree
/
Copy pathtest_kickweb.py
More file actions
76 lines (60 loc) · 3.07 KB
/
Copy pathtest_kickweb.py
File metadata and controls
76 lines (60 loc) · 3.07 KB
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
import os
import sys
import tempfile
import unittest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "irl4you"))
from kickweb import KickPublic
class Reader:
def _idle(self):
return False
def source_on(self, name):
return True
def _add(self, item):
pass
class Kick(unittest.TestCase):
def test_normalize(self):
n = KickPublic.normalize
self.assertEqual(n("XQC"), "xqc")
self.assertEqual(n("@xqc"), "xqc")
self.assertEqual(n("https://kick.com/Mein-Kanal/videos"), "mein-kanal")
self.assertEqual(n("kick.com/abc"), "abc")
self.assertEqual(n("https://evil.example/abc"), "")
self.assertEqual(n("a b"), "")
self.assertEqual(n(""), "")
def test_kanal_speichern(self):
with tempfile.TemporaryDirectory() as d:
p = os.path.join(d, "k.json")
k = KickPublic(p, Reader())
k.set_channel("kick.com/Test_1")
self.assertEqual(KickPublic(p, Reader()).channel, "test_1")
with self.assertRaises(ValueError):
k.set_channel("http://x.de/y")
def test_nachricht_mit_emotes(self):
k = KickPublic(os.path.join(tempfile.gettempdir(), "kick-test-nonexistent.json"), Reader())
ev = {"id": "abc", "content": "hi [emote:123:Kappa] da", "created_at": "2026-10-22T23:48:23+00:00",
"sender": {"username": "Bob", "identity": {"color": "#BC66FF", "badges": [{"type": "moderator"}, {"type": "subscriber"}, {"type": "x"}]}}}
(m,) = k.parse("ChatMessageEvent", ev)
self.assertEqual((m["src"], m["name"], m["text"], m["badges"], m["color"]), ("kick", "Bob", "hi Kappa da", ["moderator", "subscriber"], "#BC66FF"))
self.assertEqual(m["emotes"], [["https://files.kick.com/emotes/123/fullsize", 3, 7]])
self.assertEqual(m["ts"], 1792712903)
def test_loeschen_und_leer(self):
k = KickPublic(os.path.join(tempfile.gettempdir(), "kick-test-nonexistent.json"), Reader())
self.assertEqual(k.parse("MessageDeletedEvent", {"message": {"id": "x1"}})[0]["mid"], "x1")
self.assertEqual(k.parse("ChatMessageEvent", {"content": " ", "sender": {}}), [])
self.assertEqual(k.parse("Unbekannt", {}), [])
self.assertEqual(k.parse("ChatMessageEvent", "kaputt"), [])
def test_verlauf_und_echtzeit_ohne_doppelte(self):
got = []
class R(Reader):
def _add(self, item):
got.append(item["text"])
k = KickPublic(os.path.join(tempfile.gettempdir(), "kick-test-nonexistent.json"), R())
hist = {"id": "m1", "content": "eins", "created_at": "2026-10-23T00:00:07Z", "sender": {"username": "A"}}
k._emit(k.parse("ChatMessageEvent", hist), "hist")
k._emit(k.parse("ChatMessageEvent", dict(hist, content="eins")), "ws")
k._emit(k.parse("ChatMessageEvent", dict(hist, id="m2", content="zwei")), "ws")
self.assertEqual(got, ["eins", "zwei"])
st = k.status()
self.assertEqual((st["msgs"], st["via"]), (2, {"ws": 1, "hist": 1}))
if __name__ == "__main__":
unittest.main()